Обмін даними між завданнями

Створення конвеєрів даних з Airflow

Volker Janz

Senior Developer Advocate at Astronomer

Що таке XCom?

Схема push/pull для XCom

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

XComArg: підхід TaskFlow

Повернене @task значення стає XComArg
XComArg — це ледаче посилання, а не самі дані
Передавання його в іншу функцію створює залежність
Під час виконання Airflow розв'язує його, тягнучи з XCom

XComArgs у TaskFlow

XComArg у коді

@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-серіалізованим (dict, list, string, number)
Кожне значення додає навантаження на базу даних

Власні бекенди XCom

Маршрутизація у власний бекенд XCom

Маршрутизуйте значення до об'єктного сховища
AIRFLOW__COMMON_IO__XCOM_OBJECTSTORAGE_THRESHOLD: зберігати XCom в об'єктному сховищі лише після досягнення порога

Давайте потренуємось!

Створення конвеєрів даних з Airflow