@task def consume_two(data): return f"Got data['source']" @task def fetch_urls() -> list[str]: return ["http://a.com", "http://b.com"] @task def download(url: str) -> str: # download content return f"content_of_url"
Now go build DAGs that actually share information – cleanly and reliably. xcom in airflow
aggregate(download.expand(url=fetch_urls())) When abused, they break your DAGs
XCom (short for cross‑communication ) is Airflow’s built‑in mechanism for exchanging small pieces of data between tasks. When used wisely, they unlock powerful patterns. When abused, they break your DAGs. Let’s see how to use them correctly. XComs are key‑value pairs stored in Airflow’s metadata database. A task can push an XCom (write a value under a key), and another task can pull that value (read it). A task can push an XCom (write a
def pull_function(**context): user_id = context['ti'].xcom_pull(task_ids='push_task', key='user_id') print(f"Received user_id")
push >> pull Pattern 1: Passing an ID from a query to a processing task @task def get_latest_record_id() -> int: # Imagine a SQL query here return 42 @task def process_record(record_id: int): print(f"Processing record record_id")