Loading Academy Hub...
Master modern Airflow 2 authoring: the TaskFlow API (`@dag`, `@task`), automatic XCom argument passing, custom XCom backends (S3/ADLS), and eliminating legacy operator boilerplate.
[!NOTE] Airflow 2 introduced the TaskFlow API, allowing developers to write clean, Pythonic workflows using decorators (
@dag,@task). Return values from decorated tasks are automatically converted into XComs and injected into downstream tasks.
xcom_push and xcom_pull calls with string task IDs.@dag/@task), Data-Aware Asset Scheduling & Idempotency@asset / Dataset Lineage): Instead of brittle ExternalTaskSensor polling timers (02:00 AM wait for 01:45 AM ingestion), declare Data Assets (outlets=[Asset("s3://lake/silver/orders")]). Downstream Gold/ML DAGs trigger schedule=[orders_silver_asset] automatically the instant upstream commits succeed!data_interval_start) vs datetime.now(): Never call datetime.now() inside SQL queries or S3 output paths—backfilling last Tuesday would overwrite today’s partition! Always pass {{ data_interval_start }} / TaskFlow context variables (logical_date) and write outputs idempotently (INSERT OVERWRITE / partition overwrite / MERGE).< 48 KB Metadata Rule): TaskFlow @task return values serialize automatically via XCom. Pass S3/ADLS/GCS URIs or table names between tasks—never return raw DataFrames or megabyte JSON payloads unless backed by an Object Storage XCom Backend.Review the production code, company interview Q&As, incident traces, and comprehension quiz inside this lesson.
from airflow.decorators import dag, task
from datetime import datetime
@dag(
dag_id='taskflow_clean_orders',
schedule='@daily',
start_date=datetime(2026, 1, 1),
catchup=False
)
def taskflow_etl_pipeline():
@task
def extract_order_batch() -> dict:
return {"batch_id": "b_20260301", "record_count": 50000}
@task
def validate_and_transform(metadata: dict) -> str:
print(f"Validating batch: {metadata['batch_id']} with {metadata['record_count']} records.")
return f"s3://lakehouse/silver/batch_{metadata['batch_id']}.parquet"
@task
def load_to_warehouse(target_s3_path: str):
print(f"Executing COPY INTO Snowflake from path: {target_s3_path}")
# Dynamic dataflow dependency wiring
batch_meta = extract_order_batch()
transformed_path = validate_and_transform(batch_meta)
load_to_warehouse(transformed_path)
# Instantiate DAG
taskflow_etl_pipeline()sqlalchemy.exc.DataError: (psycopg2.errors.StringDataRightTruncation) value too long for type character varying.