10 Lines of YAML vs 200 Lines of Airflow DAG

version: "3.8"
services:
  postgres:
    image: postgres:15
    environment:
      POSTGRES_USER: airflow
      POSTGRES_PASSWORD: airflow
      POSTGRES_DB: airflow
    volumes:
      - postgres-data:/var/lib/postgresql/data

  webserver:
    image: apache/airflow:2.8.0
    command: webserver
    ports:
      - "8080:8080"
    environment:
      AIRFLOW__CORE__EXECUTOR: LocalExecutor
      AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow
      AIRFLOW__CORE__FERNET_KEY: "your-fernet-key-here"
    volumes:
      - ./dags:/opt/airflow/dags
      - ./plugins:/opt/airflow/plugins
    depends_on:
      - postgres

  scheduler:
    image: apache/airflow:2.8.0
    command: scheduler
    environment:
      AIRFLOW__CORE__EXECUTOR: LocalExecutor
      AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow
      AIRFLOW__CORE__FERNET_KEY: "your-fernet-key-here"
    volumes:
      - ./dags:/opt/airflow/dags
      - ./plugins:/opt/airflow/plugins
    depends_on:
      - postgres
      - webserver

volumes:
  postgres-data:

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.amazon.aws.hooks.s3 import S3Hook
from airflow.models import Variable
from datetime import datetime, timedelta
import requests
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
import json
import io


default_args = {
    "owner": "data-engineering",
    "depends_on_past": True,
    "email_on_failure": True,
    "email": ["oncall@company.com"],
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
}


def extract_payments(**context):
    """Extract transactions from payments API."""
    api_key = Variable.get("payments_api_key")
    base_url = "https://api.payments.example.com/v1/transactions"

    # Get bookmark (last successful extraction timestamp)
    try:
        bookmark = Variable.get("payments_last_timestamp")
    except KeyError:
        bookmark = "2024-01-01T00:00:00Z"

    headers = {"Authorization": f"Bearer {api_key}"}
    params = {
        "created_after": bookmark,
        "limit": 1000,
    }

    all_records = []
    page = 1

    while True:
        params["page"] = page
        response = requests.get(base_url, headers=headers, params=params)
        response.raise_for_status()
        data = response.json()

        records = data.get("transactions", [])
        if not records:
            break

        all_records.extend(records)
        page += 1

        if not data.get("has_more", False):
            break

    if not all_records:
        return "No new data"

    # Push to XCom for next task
    context["ti"].xcom_push(key="raw_data", value=json.dumps(all_records))
    context["ti"].xcom_push(key="record_count", value=len(all_records))

    return f"Extracted {len(all_records)} records"


def transform_payments(**context):
    """Transform: rename columns, filter nulls, cast types."""
    raw_json = context["ti"].xcom_pull(key="raw_data", task_ids="extract")
    records = json.loads(raw_json)

    df = pd.DataFrame(records)

    # Rename columns
    df = df.rename(columns={
        "txn_id": "transaction_id",
        "amt": "amount",
        "ccy": "currency",
        "ts": "created_at",
        "cust_id": "customer_id",
    })

    # Filter nulls
    df = df.dropna(subset=["transaction_id", "amount", "customer_id"])

    # Cast types
    df["amount"] = df["amount"].astype(float)
    df["created_at"] = pd.to_datetime(df["created_at"])
    df["customer_id"] = df["customer_id"].astype(str)

    context["ti"].xcom_push(key="transformed_data", value=df.to_json())
    return f"Transformed {len(df)} records"


def load_to_s3(**context):
    """Load transformed data to S3 as Parquet."""
    df_json = context["ti"].xcom_pull(key="transformed_data", task_ids="transform")
    df = pd.read_json(df_json)

    table = pa.Table.from_pandas(df)
    buf = io.BytesIO()
    pq.write_table(table, buf)
    buf.seek(0)

    execution_date = context["execution_date"].strftime("%Y%m%d_%H%M%S")
    key = f"raw/payments/transactions/{execution_date}.parquet"

    s3_hook = S3Hook(aws_conn_id="s3_data_lake")
    s3_hook.load_bytes(
        bytes_data=buf.getvalue(),
        key=key,
        bucket_name="company-data-lake",
        replace=True,
    )

    # Update bookmark
    max_timestamp = df["created_at"].max().isoformat()
    Variable.set("payments_last_timestamp", max_timestamp)

    return f"Loaded {len(df)} rows to s3://{key}"


with DAG(
    "payments_etl",
    default_args=default_args,
    schedule_interval="@hourly",
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=["etl", "payments"],
) as dag:

    extract = PythonOperator(
        task_id="extract",
        python_callable=extract_payments,
    )

    transform = PythonOperator(
        task_id="transform",
        python_callable=transform_payments,
    )

    load = PythonOperator(
        task_id="load",
        python_callable=load_to_s3,
    )

    extract >> transform >> load

pipeline: api-crm
source:
  type: rest_api
  url: "https://api.crm.example.com/v2/contacts"
  auth: bearer ${CRM_API_KEY}
  pagination: offset
  incremental:
    field: updated_at
    strategy: bookmark

target:
  type: s3
  bucket: s3://company-data-lake
  prefix: raw/crm