ETL Pipelines Cheat Sheet

from airflow import DAGfrom airflow.operators.python import PythonOperatorfrom datetime import datetimedef extract(): ...def transform(): ...def load(): ...with DAG(    dag_id="daily_sales_etl",    schedule="0 2 * * *",       # 2am daily    start_date=datetime(2024, 1, 1),    catchup=False,) as dag:    extract_task = PythonOperator(task_id="extract", python_callable=extract)    transform_task = PythonOperator(task_id="transform", python_callable=transform)    load_task = PythonOperator(task_id="load", python_callable=load)    extract_task >> transform_task >> load_task

import pandas as pd# Extractdf = pd.read_csv("raw_sales.csv")# Transformdf["order_date"] = pd.to_datetime(df["order_date"])df = df.dropna(subset=["customer_id"])df["amount"] = df["amount"].clip(lower=0)df["region"] = df["region"].str.upper().str.strip()# Loaddf.to_sql("sales_clean", con=engine, if_exists="append", index=False)

-- models/fct_orders.sql{{  config(    materialized='incremental',    unique_key='order_id',    incremental_strategy='merge'  )}}SELECT  order_id,  customer_id,  amount,  updated_atFROM {{ source('raw', 'orders') }}{% if is_incremental() %}WHERE updated_at > (SELECT MAX(updated_at) FROM {{ this }}){% endif %}

-- Close out changed current rowsMERGE INTO dim_customer AS targetUSING staging_customer AS sourceON target.customer_id = source.customer_id AND target.is_current = TRUEWHEN MATCHED AND (target.email <> source.email OR target.address <> source.address) THEN  UPDATE SET target.is_current = FALSE, target.valid_to = CURRENT_DATEWHEN NOT MATCHED THEN  INSERT (customer_id, email, address, valid_from, valid_to, is_current)  VALUES (source.customer_id, source.email, source.address, CURRENT_DATE, NULL, TRUE);-- Insert the new current row for every record that was just closed outINSERT INTO dim_customer (customer_id, email, address, valid_from, valid_to, is_current)SELECT s.customer_id, s.email, s.address, CURRENT_DATE, NULL, TRUEFROM staging_customer sJOIN dim_customer d  ON d.customer_id = s.customer_id AND d.is_current = FALSE AND d.valid_to = CURRENT_DATE;

from airflow.sensors.external_task import ExternalTaskSensorfrom airflow.decorators import taskwait_for_upstream = ExternalTaskSensor(    task_id="wait_for_upstream_dag",    external_dag_id="raw_ingestion",    external_task_id="load_raw",    timeout=3600,    mode="reschedule",   # frees the worker slot while waiting, unlike mode="poke")@taskdef get_source_tables():    return ["orders", "customers", "products"]@taskdef load_table(table_name: str):    ...  # extract + load a single table# Creates one mapped task instance per table, all running in parallelwait_for_upstream >> load_table.expand(table_name=get_source_tables())