undefined

---

-- Document source tables
SELECT
    table_name,
    column_name,
    data_type,
    is_nullable
FROM information_schema.columns
WHERE table_schema = 'source_schema'
ORDER BY table_name, ordinal_position;

python scripts/pipeline_orchestrator.py generate \
  --type airflow \
  --source postgres \
  --tables orders,customers,products \
  --mode incremental \
  --watermark updated_at \
  --output dags/extract_source.py

-- models/staging/stg_orders.sql
WITH source AS (
    SELECT * FROM {{ source('postgres', 'orders') }}
),

renamed AS (
    SELECT
        order_id,
        customer_id,
        order_date,
        total_amount,
        status,
        _extracted_at
    FROM source
    WHERE order_date >= DATEADD(day, -3, CURRENT_DATE)
)

SELECT * FROM renamed

-- models/marts/fct_orders.sql
{{
    config(
        materialized='incremental',
        unique_key='order_id',
        cluster_by=['order_date']
    )
}}

SELECT
    o.order_id,
    o.customer_id,
    c.customer_segment,
    o.order_date,
    o.total_amount,
    o.status
FROM {{ ref('stg_orders') }} o
LEFT JOIN {{ ref('dim_customers') }} c
    ON o.customer_id = c.customer_id

{% if is_incremental() %}
WHERE o._extracted_at > (SELECT MAX(_extracted_at) FROM {{ this }})
{% endif %}

undefined

undefined

undefined

extract = BashOperator(
    task_id='extract_source_data',
    bash_command='python /opt/airflow/scripts/extract.py --date {{ ds }}',
)

transform = BashOperator(
    task_id='run_dbt_models',
    bash_command='cd /opt/airflow/dbt && dbt run --select marts.*',
)

test = BashOperator(
    task_id='run_dbt_tests',
    bash_command='cd /opt/airflow/dbt && dbt test --select marts.*',
)

notify = BashOperator(
    task_id='send_notification',
    bash_command='python /opt/airflow/scripts/notify.py --status success',
    trigger_rule='all_success',
)

extract >> transform >> test >> notify

undefined

undefined

---

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "title": "UserEvent",
  "type": "object",
  "required": ["event_id", "user_id", "event_type", "timestamp"],
  "properties": {
    "event_id": {"type": "string", "format": "uuid"},
    "user_id": {"type": "string"},
    "event_type": {"type": "string", "enum": ["page_view", "click", "purchase"]},
    "timestamp": {"type": "string", "format": "date-time"},
    "properties": {"type": "object"}
  }
}

undefined

undefined

undefined

undefined

undefined

    # Write valid records
    valid_df.write \
        .format("delta") \
        .mode("append") \
        .save("/data/lake/user_events")

    # Write invalid to DLQ
    if invalid_df.count() > 0:
        invalid_df \
            .withColumn("error_timestamp", current_timestamp()) \
            .withColumn("error_reason", lit("missing_event_id")) \
            .write \
            .format("delta") \
            .mode("append") \
            .save("/data/lake/dlq/user_events")

except Exception as e:
    # Log error, alert, continue
    logger.error(f"Batch {batch_id} failed: {e}")
    raise

undefined

undefined

    if progress['sources']:
        # Calculate lag from latest offset
        for source in progress['sources']:
            end_offset = source.get('endOffset', {})
            # Parse Kafka offsets and calculate lag

---

undefined

undefined

undefined

undefined

undefined

undefined

undefined

undefined

undefined

report = {
    "table": table_name,
    "timestamp": datetime.now().isoformat(),
    "checks": {}
}

# Row count check
row_count = connection.execute(
    f"SELECT COUNT(*) FROM {table_name}"
).fetchone()[0]
report["checks"]["row_count"] = {
    "value": row_count,
    "status": "pass" if row_count > 0 else "fail"
}

# Freshness check
max_date = connection.execute(
    f"SELECT MAX(created_at) FROM {table_name}"
).fetchone()[0]
hours_old = (datetime.now() - max_date).total_seconds() / 3600
report["checks"]["freshness"] = {
    "max_timestamp": max_date.isoformat(),
    "hours_old": round(hours_old, 2),
    "status": "pass" if hours_old < 24 else "fail"
}

# Null rate check
null_query = f"""
SELECT
    SUM(CASE WHEN order_id IS NULL THEN 1 ELSE 0 END) as null_order_id,
    SUM(CASE WHEN customer_id IS NULL THEN 1 ELSE 0 END) as null_customer_id,
    COUNT(*) as total
FROM {table_name}
"""
null_result = connection.execute(null_query).fetchone()
report["checks"]["null_rates"] = {
    "order_id": null_result[0] / null_result[2] if null_result[2] > 0 else 0,
    "customer_id": null_result[1] / null_result[2] if null_result[2] > 0 else 0,
    "status": "pass" if null_result[0] == 0 and null_result[1] == 0 else "fail"
}

# Duplicate check
dup_query = f"""
SELECT COUNT(*) - COUNT(DISTINCT order_id) as duplicates
FROM {table_name}
"""
duplicates = connection.execute(dup_query).fetchone()[0]
report["checks"]["duplicates"] = {
    "count": duplicates,
    "status": "pass" if duplicates == 0 else "fail"
}

# Overall status
all_passed = all(
    check["status"] == "pass"
    for check in report["checks"].values()
)
report["overall_status"] = "pass" if all_passed else "fail"

return report

---

Is real-time insight required?
├── Yes → Use streaming
│   └── Is exactly-once semantics needed?
│       ├── Yes → Kafka + Flink/Spark Structured Streaming
│       └── No → Kafka + consumer groups
└── No → Use batch
    └── Is data volume > 1TB daily?
        ├── Yes → Spark/Databricks
        └── No → dbt + warehouse compute

Task exceeded max execution time

undefined

---

**Symptom:** Spark job OOM

**Solution:**
1. Increase executor memory
2. Reduce partition size
3. Use disk spill
```python
spark.conf.set("spark.executor.memory", "8g")
spark.conf.set("spark.sql.shuffle.partitions", "200")
spark.conf.set("spark.memory.fraction", "0.8")

Consumer lag: 1000000 messages

undefined

---

Expected unique, found 150 duplicates

-- dbt incremental with dedup
{{
    config(
        materialized='incremental',
        unique_key='order_id'
    )
}}

SELECT * FROM (
    SELECT
        *,
        ROW_NUMBER() OVER (
            PARTITION BY order_id
            ORDER BY updated_at DESC
        ) as rn
    FROM {{ source('raw', 'orders') }}
) WHERE rn = 1

Last update: 3 days ago

undefined

---

**Symptom:** Schema drift detected

**Solution:**
1. Update data contract
2. Modify transformations
3. Communicate with producers
```python

---

Query runtime: 4 hours (expected: 30 minutes)

-- Before: Full table scan
SELECT * FROM orders WHERE order_date = '2024-01-15';

-- After: Partition pruning
-- Table partitioned by order_date
SELECT * FROM orders WHERE order_date = '2024-01-15';

-- Add clustering for frequent filters
ALTER TABLE orders CLUSTER BY (customer_id);

Model fct_orders completed in 45 minutes

-- Convert to incremental
{{
    config(
        materialized='incremental',
        unique_key='order_id',
        on_schema_change='sync_all_columns'
    )
}}

SELECT * FROM {{ ref('stg_orders') }}
{% if is_incremental() %}
WHERE _loaded_at > (SELECT MAX(_loaded_at) FROM {{ this }})
{% endif %}