airflow xcom: как передавать данные между задачами DAG

Airflow XCom передаёт небольшой сериализуемый результат одной задачи DAG другой задаче в рамках запуска: получатель читает значение по источнику и ключу. Для рабочего контура задайте формат результата, доступ к внешним данным и поведение при пустом значении; крупный файл храните отдельно, а через XCom передавайте ссылку. Такой маршрут подходит для метаданных, идентификаторов и статусов, которые помещаются в небольшой объект.

Границы XCom

В Airflow задачи могут выполняться на разных машинах, поэтому локальная переменная первой задачи второй задаче недоступна. XCom даёт штатный канал обмена: запись привязана к задаче, DAG, ключу и запуску. Задача-получатель указывает источник явно. Для функции с декоратором @task результат обычно попадает под ключ return_value, а TaskFlow передаёт его следующей функции как аргумент. При ручной передаче через контекст задачи используют xcom_push и xcom_pull. Эти механизмы решают один вопрос — доставку малого значения, а смысл значения задаёт команда.

Возьмём условный процесс: задача выгружает реестр заказов в разрешённое хранилище, следующая запускает проверку качества, третья готовит сводку для сотрудника. Через XCom первая задача возвращает путь, версию схемы и число строк. Получателю достаточно этой карточки, чтобы найти файл и сверить ожидаемый формат. Сам реестр остаётся там, где настроены права, срок хранения и журнал обращений. Посмотрите разбор операторов для ИИ-шага Airflow: здесь акцент уже на контракте между соседними задачами.

Если отдать в XCom весь датафрейм, размер записей и нагрузка на служебное хранилище растут вместе с объёмом данных. Официальная документация Airflow прямо относит XCom к обмену малыми значениями. Хранилище выбирают по требованиям проекта, адрес передают получателю с доступом. Адрес без права чтения бесполезен, поэтому доступ проверяют отдельно от факта успешной передачи XCom.

Контракт результата

Прежде чем писать код, согласуйте между владельцами задач контракт сообщения. Запишите task_id производителя, имя ключа, тип значения, обязательные поля и версию схемы. Например, для карточки выгрузки можно выбрать строковые поля uri и schema_version, целое поле row_count и строковый checksum. Это проектная схема: имена полей и формат суммы утверждает команда. Версия описывает ваш формат данных; версия Airflow фиксируется отдельно.

Сервер приложения или хранилище проверяет полномочия учётной записи получателя перед чтением артефакта. Право чтения файла проверяет хранилище независимо от наличия адреса в XCom. Секреты, токены и клиентские данные в карточку результата помещать нельзя: для них используют предназначенный проектом механизм доступа и правила обработки данных. Если последующая задача обращается к сервису с ИИ, модель может подготовить черновик вывода, а спорные выводы подтверждает ответственный человек. Архитектуру доступа к документам подробнее разбирает разбор Airflow connections и доступа к источникам.

Схему удобно держать рядом с кодом DAG и изменять через обычное ревью. При добавлении обязательного поля повышайте версию контракта и обновляйте получателя согласованно. Иначе успешный запуск производителя способен передать сообщение, которое следующая задача поймёт неверно. Для необязательных полей задайте значения по умолчанию и явно опишите, когда они допустимы.

Маршрут передачи

Постройте передачу от потребности получателя. Сначала определите минимальный набор полей, затем место хранения тяжёлого результата и учётную запись читателя. Для небольшого идентификатора TaskFlow удобен тем, что связывает возвращённое значение с аргументом следующей задачи. Для явного ключа в задаче получите контекст и вызовите task_instance.xcom_push(key="artifact_ref", value=card); получатель вызовет task_instance.xcom_pull(task_ids="export_orders", key="artifact_ref"). Имена здесь условные, а синтаксис следует сверять с документацией установленной версии Airflow.

1. Производитель сохраняет большой файл в разрешённом хранилище и получает адрес, доступный только нужной роли.
2. Производитель собирает короткую карточку, проверяет её собственные обязательные поля и записывает значение в XCom.
3. Получатель явно указывает task_ids и ключ, проверяет тип, версию и наличие ожидаемых полей.
4. Получатель обращается к хранилищу от своей учётной записи; сервер проверяет полномочия и возвращает файл либо отказ.
5. После обработки получатель фиксирует статус, причину исключения и ссылку на журнал запуска для приёмки.

Важно различать порядок выполнения задач и смысл сообщения. Зависимость в DAG заставляет получателя дождаться производителя, а качество артефакта требует отдельной проверки. Добавьте прикладные проверки после чтения: ожидаемый формат, совпадение контрольной суммы, допустимость пустого набора и соответствие версии. При отказе доступа задача должна завершиться понятной ошибкой, без подстановки старого файла из другой выгрузки. Для проектирования соседних автоматизированных шагов пригодится разбор повторов задач Airflow; сам XCom остаётся механизмом Airflow.

Переносить этот маршрут на весь контур удобнее после проверки одной пары задач и одного типа результата. Если ваш DAG уже возвращает сложный объект, выделите короткую карточку и её владельца.

Исключения и повторы

При пустом XCom получатель сперва проверяет, завершился ли производитель успешно, указан ли правильный task_ids и совпадает ли ключ. В актуальной документации Airflow вызов xcom_pull без task_ids ищет значение текущей задачи; при чтении соседней задачи источник лучше указывать всегда. Автоматическая запись результата зависит от поведения оператора и настройки do_xcom_push. Поэтому перед запуском рабочего DAG команда проверяет, откуда именно приходит return_value и требуется ли собственный ключ.

Повтор задачи меняет состояние обмена. Airflow очищает XCom задачи при повторной попытке после сбоя, чтобы запуск оставался идемпотентным. Следовательно, карточка XCom годится только для обмена в рамках запуска; долговременное состояние и прогресс между попытками храните отдельно. Если внешняя выгрузка уже создана, повтор должен безопасно найти её по идентификатору запуска либо создать новую по согласованному правилу. Удаление старого файла, замена ссылки и срок хранения требуют отдельного решения владельца данных.

• Пустое значение: отметьте, допустим ли пустой результат по контракту; иначе завершите задачу с объяснением.
• Чужая версия: остановите ветку и отправьте владельцу DAG номер полученной версии и ожидаемый формат.
• Крупный результат: сохраните файл отдельно; через XCom передайте короткую ссылку и контрольные поля.
• Отказ в доступе: проверьте разрешения учётной записи на сервере хранилища, без расширения прав для всех задач.

Если DAG вызывает другой DAG, область поиска XCom надо задавать особенно аккуратно: официальная документация показывает указание dag_id и run_id для конкретного запуска. Внутри обычной цепочки одного DAG достаточно явного источника и ключа, а внешние переходы лучше описывать отдельным контрактом. Ошибка в идентификаторе запуска может вернуть пустоту или чужой по смыслу результат, поэтому тест на изоляцию запусков включают в приёмку.

Приёмка обмена

Приёмка начинается с воспроизводимого тестового набора без реальных клиентских данных. Для производителя зафиксируйте ожидаемую карточку; для получателя — проверку типов, версии, адреса и права чтения. Проведите запуск с корректным файлом, пустым ответом, устаревшей схемой, ошибочной контрольной суммой и отказом хранилища. По каждому варианту ожидается наблюдаемый итог: обработка, остановка ветки либо передача на ручной разбор. Один успешный зелёный запуск подтверждает только счастливый путь.

В журнале оставляйте идентификатор DAG и запуска, задачу-источник, ключ XCom, версию схемы и код исключения. Полный файл, секреты и персональные данные в журнал выводить нельзя. Ответственный за процесс просматривает результаты спорных проверок и подтверждает выводы, особенно при обработке моделью ИИ. Сервер проверяет полномочия до доступа к артефакту; модель готовит лишь черновик; полномочия и окончательное решение о качестве данных остаются за сервером и ответственным человеком.

После теста закрепите владельцев: разработчик DAG отвечает за контракт XCom, администратор хранилища — за права и срок хранения, владелец процесса — за смысл результата и правила ручной проверки. Изменение формата оформляйте как изменение схемы с проверкой старого и нового потребителя. Для широкой схемы автоматизации, где несколько систем обмениваются результатами, можно обратиться к проектированию автоматизации бизнес-процессов: состав работ и стоимость обсуждаются под конкретные источники и ограничения.

Частые вопросы