¿Cómo trabajó con los archivos al reiniciar el DAG después de que falló, para no sobrescribir los datos en S3 con datos incompletos?