Обмен данными между задачами

Построение пайплайнов данных с Airflow

Volker Janz

Senior Developer Advocate at Astronomer

Что такое XCom?

Схема push/pull XCom

XCom = межзадачная коммуникация
Значения хранятся в базе метаданных
Классический подход: явные вызовы xcom_push() и xcom_pull()

XComArgs: подход TaskFlow

Возвращаемое значение @task становится XComArg
XComArg — это отложенная ссылка, а не сами данные
Передача в другую функцию создаёт зависимость
Во время выполнения Airflow разрешает её, извлекая данные из XCom

XComArgs в TaskFlow

XComArgs в коде

@task
def get_config():
    return {"name": "daily_etl"}

@task
def run_pipeline(config):
    print(config["name"])


config = get_config()   # XComArg

run_pipeline(config)    # dependency + data

Совместное использование классического подхода и TaskFlow

get_timestamp = BashOperator(
    task_id="get_timestamp",
    bash_command="date +%Y-%m-%d")

@task
def log_timestamp(ts_value):
    print(f"Timestamp: {ts_value}")


log_timestamp(get_timestamp.output)

Ограничения XCom

Ограничения размера XCom по типу базы данных

Максимальный размер зависит от базы данных
Значения должны быть сериализуемы в JSON (словари, списки, строки, числа)
Каждое значение создаёт нагрузку на базу данных

Пользовательские XCom backends

Маршрутизация через пользовательский XCom backend

Направляйте значения в объектное хранилище
AIRFLOW__COMMON_IO__XCOM_OBJECTSTORAGE_THRESHOLD: XCom сохраняется в объектное хранилище только при достижении порогового значения

Давайте потренируемся!

Построение пайплайнов данных с Airflow