Воздушный поток: невозможно отправить данные обратно в конвейер
Я настроил рабочий процесс, который состоит из двух задач:
Во-вторых, я успешно извлекаю данные, отправленные в конвейер первой задачей. Однако после обработки данных, когда я пытаюсь отправить данные обратно в конвейер, я получаю сообщение об ошибке «ИНФОРМАЦИЯ — Задача завершена с кодом возврата -9».
Вот логи:
Почему XCOM не может отправить данные в конвейер? Как я могу передавать данные другим задачам?
Ответы 2
Код возврата -9 чаще всего связан с ошибкой нехватки памяти. Например, насколько велик размер ваших данных по сравнению с рабочей памятью, выполняющей задачу «primary_transform_task»? Избавьтесь от неиспользуемых переменных и оптимизируйте память или приобретите рабочий узел большего размера для размещения больших данных. Также помните, что лучшее использование воздушного потока — это оркестровка. Например, обработка больших наборов данных должна осуществляться в кластерах Spark.
Проверьте, какой сервер XCOM вы используете. Примечание. XCom идентифицируется ключом (по сути, его именем), а также идентификатором Task_id и dag_id, из которых он получен. Они могут иметь любое (сериализуемое) значение, но предназначены только для небольших объемов данных; не используйте их для передачи больших значений, например кадров данных. Ссылка: airflow.apache.org/docs/apache-airflow/stable/core-concepts/…
Да, понял... Но у меня есть два вопроса.... Первый... Как я могу передавать данные между двумя задачами, если не использую XCOM.... второй.... Почему я могу передавать/извлекать данные в/из задачи 1, но нажатие вызывает ошибку в задаче 2
Выявленная проблема:
Проблема была в перекрестной связи, то есть XCOM, которая не подходит для передачи кадров данных, и больших объемах данных в конвейерах. Ссылка: https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/xcoms.html
Решение:
Поскольку мы не можем передавать данные между задачами с помощью XCOM, мы можем использовать общее хранилище, с помощью которого задачи могут передавать данные. XCOM можно использовать для передачи местоположения файлов в общем хранилище, чтобы другие задачи могли знать, откуда получать данные.
Данные, которые я читаю в задаче 1 и отправляю в конвейер, имеют размер около 777 МБ, и я могу их вытащить. После очистки данных в основной задаче размер данных становится 333 МБ. Вставка его в конвейер дает ошибку