Как вы работали с файлами при повторном запуске DAG после его падения, чтобы не перезаписать данные в S3 неполными данными?