Большой CSV между задачами Airflow: ссылка в XCom вместо содержимого

Задача загрузки возвращает большой список строк. TaskFlow автоматически передаёт результат следующей задаче через XCom, и база метаданных начинает хранить содержимое выгрузок. При переносе на несколько workers появляется ещё одна ошибка: путь /tmp/export.csv существует только на машине производителя.

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

Минимальный DAG с общим S3-хранилищем

Пример для Airflow 3.x требует boto3 на workers и переменную DATA_BUCKET с именем подготовленного бакета. Доступ S3 задаётся стандартной конфигурацией роли или окружения. Производитель формирует маленький учебный CSV; в рабочем загрузчике на его месте будет чтение источника и запись подготовленного файла.

from datetime import datetime, timezone
import csv
import hashlib
import io
import os

import boto3
from airflow.sdk import dag, task, get_current_context

@dag(schedule=None, start_date=datetime(2026, 1, 1, tzinfo=timezone.utc), catchup=False)
def file_reference_demo():
    @task
    def produce():
        context = get_current_context()
        run_key = hashlib.sha256(context["run_id"].encode()).hexdigest()
        body = b"id,qty\nA,2\nB,3\n"
        bucket = os.environ["DATA_BUCKET"]
        key = f"demo/{run_key}/input.csv"
        boto3.client("s3").put_object(Bucket=bucket, Key=key, Body=body)
        return {"bucket": bucket, "key": key, "size": len(body),
                "sha256": hashlib.sha256(body).hexdigest()}

    @task
    def consume(reference):
        response = boto3.client("s3").get_object(
            Bucket=reference["bucket"], Key=reference["key"])
        stream = response["Body"]
        try:
            body = stream.read()
        finally:
            stream.close()
        if len(body) != reference["size"]:
            raise ValueError("Object size changed")
        if hashlib.sha256(body).hexdigest() != reference["sha256"]:
            raise ValueError("Object content changed")
        rows = csv.DictReader(io.StringIO(body.decode("utf-8")))
        total = sum(int(row["qty"]) for row in rows)
        print("total_qty", total)

    consume(produce())

file_reference_demo()

После ручного запуска сумма равна пяти, а XCom содержит только четыре небольших поля. Запись объекта заканчивается до возврата ссылки. Если сохранение завершится ошибкой, downstream не получит сообщение о готовом файле.

Для большого объекта чтение целиком нужно заменить потоковым разбором или обработкой средствами Spark. В примере память расходуется на весь маленький CSV, чтобы показать передачу и проверку содержимого без стороннего формата. Принцип ссылки от размера не меняется.

Ссылка должна переживать повторный запуск

Ключ включает хеш run_id, поэтому соседние запуски не пишут в один общий export.csv. Повтор produce в том же запуске использует прежний ключ. Для изменяемого реального источника выбирайте неизменяемое имя по версии входа, S3 VersionId либо отдельному идентификатору попытки. Иначе retry производителя может заменить файл, который уже начала читать другая задача.

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

Проверьте доступ с обеих сторон

Производителю нужны права записи, потребителю — чтения нужного префикса. Локальный удачный запуск на одном worker не проверяет распределённую доступность. Выполните задачи в реальном executor и убедитесь, что принимающая роль открывает объект без передачи секретов через XCom.

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

В Data Engineering Fast-Track вы соедините Airflow, Kafka, Spark и lakehouse в итоговом конвейере. Откройте программу и научитесь передавать между этапами устойчивые результаты, которые можно проверить и переобработать.