Как передавать большие данные между тасками в Airflow

Что проверяют

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

Как отвечать

• почему не XCom: значение сериализуется и пишется в таблицу метабазы. Жёсткий предел задаёт тип колонки, он зависит от СУБД и версии Airflow, но проблемы начинаются раньше: база разбухает, интерфейс тормозит, данные идут через метабазу на каждом запуске;
• ссылка вместо данных: таск пишет результат в S3, GCS или HDFS, обычно в Parquet, и возвращает путь. В TaskFlow API это обычный return строки, следующий таск получает её аргументом. Путь делают детерминированным, с run_id внутри, тогда перезапуск перезапишет свой файл и не создаст дубль;
• кастомный XCom backend: наследник BaseXCom с serialize_value и deserialize_value, который сам кладёт объект в хранилище и сохраняет в базе только ссылку. Готовый вариант есть в провайдере common.io: XComObjectStorageBackend с порогом размера, выше которого значение уходит в объектное хранилище;
• не гонять данные через воркер: если данные уже в хранилище, тяжёлую работу пусть делают Spark, dbt или сама СУБД, а Airflow только оркеструет. Пример: INSERT ... SELECT внутри хранилища вместо выгрузки в pandas;
• локальный диск: работает, только если все таски гарантированно идут на одной машине. В CeleryExecutor и KubernetesExecutor это не так.

Где спотыкаются

Называют фиксированный лимит XCom как константу Airflow. Предел задаёт колонка в метабазе, и на разных СУБД он разный.

Предлагают поднять parallelism или сменить executor. Это про число одновременных тасков, к передаче данных отношения не имеет.

Возвращают из таска pandas DataFrame и радуются, что на тестовых данных всё работает. На проде это сотни мегабайт в метабазе на каждый запуск DAG.