Compartilhando dados entre tarefas

Criando Pipelines de Dados com Airflow

Volker Janz

Senior Developer Advocate at Astronomer

O que é XCom?

Fluxo de push/pull do XCom

XCom = comunicação entre tarefas
Valores salvos no banco de metadados
Abordagem clássica: xcom_push() e xcom_pull() explícitos

XComArgs: o jeito TaskFlow

O retorno de @task vira um XComArg
XComArg é uma referência preguiçosa, não o dado em si
Passá-lo a outra função cria a dependência
Em runtime, o Airflow resolve puxando do XCom

XComArgs no TaskFlow

XComArgs no código

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

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


config = get_config()   # XComArg

run_pipeline(config)    # dependência + dados

Misturando clássico e 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)

Limitações do XCom

Limites de tamanho do XCom por backend

Tamanho máximo depende do seu backend de banco de dados
Deve ser serializável em JSON (dicts, listas, strings, números)
Cada valor adiciona carga ao banco

Backends XCom personalizados

Roteamento com backend XCom personalizado

Direcione valores para um armazenamento de objetos
AIRFLOW__COMMON_IO__XCOM_OBJECTSTORAGE_THRESHOLD: só armazena XComs no objeto quando o limite é atingido

Vamos praticar!

Criando Pipelines de Dados com Airflow