Привет!

Столкнулась с проблемой:

В таске функция обрабатывает файлы в цикле. Логика такая:

1. Копируем файл в папку, к которой подцеплена таблица, на hdfs
2. Из таблицы вставляем обработанные данные в целевую таблицу hive
3. Удаляем файл

Копирование и удаление через ssh hook, вставка - jdbc

В рандомные моменты даг падает с записью:

WARNING - --- Logging error ---
{{local_task_job.py:103}} INFO - Task exited with return code 1

У меня вообще нет идей, что это может быть, так как падает всегда в разных местах

Есть идеи по этому поводу?)

сеть шалит?

скорее всего на вставке в джбс

Там еще исход разный:

Иногда остается папка и файл в ней, а иногда папки нет, хотя я ее вроде бы и не удаляю

Может ли быть такое, что удаление файла не всегда дожидается окончания инсерта, если оно записано вот так:

       conn.exec_command(copy_command)
       print(copy_command)
       hook.run(insert)
       conn.exec_command(remove_command)
       print (remove_command)
       print('This file was loaded and removed: ', file)

А зачем?)

это все в одном операторе выполняется?

Да

Так может имеет смысл разделить на разные операторы и добавить условие тригера? Или же сенсор на таблицу после инсерта

Продублирую)
Вдруг кто появится

Я не понимаю как в ацикличном графе реализовать цикл по файлам на уровне операторов (

Вес таскам проставляй

И будет тебе счастье

Там можно у сенсора указать любое время даграна другого дага же

У меня сейчас запускается и смотрит 3 месяца прошлого квартала

Не очень поняла идею... Это помогает распараллелить выполнение на несколько даг ранов, чтобы последующие таски ждали выполнения предыдущих из каждого запуска. Так? Тем не менее это не отвечает на вопрос о цикле, если каждый файл должен обрабатываться последовательно

в цикле генерить операторы где каждый следующий зависит от предыдущего

я выполняю, если нужно вручную триггернуть, просто мечу первую такску как входную точку внитри самого граф вью как саксесс

можно, но это же код надо править

как вариант, спасибо