---
description: Pipeline Design
alwaysApply: false
---

# Pipeline Design

Patterns for building reliable data pipelines.

## Core Principles

- **Idempotency** — same inputs, same outputs, every time
- **Determinism** — use `execution_date`, not `current_timestamp()`; seed random operations
- **Atomicity** — all-or-nothing writes; use Delta/Iceberg transactions or staging+swap

## Pipeline Patterns

### Batch Full Refresh
Use when source is small or doesn't support incremental.

```python
df = spark.read.table(source)
df.write.mode("overwrite").saveAsTable(target)
```

### Batch Incremental (Watermark)
Use when data is large and source supports change tracking.

```python
last_wm = get_watermark(target, wm_col)
new = spark.read.table(source).filter(F.col(wm_col) > last_wm)
DeltaTable.forName(spark, target).alias("t") \
    .merge(new.alias("s"), "t.id = s.id") \
    .whenMatchedUpdateAll().whenNotMatchedInsertAll().execute()
```

### CDC (Change Data Capture)
Use when tracking all changes for point-in-time queries.

```python
DeltaTable.forName(spark, target).alias("t") \
    .merge(cdc_events.alias("s"), "t.id = s.id") \
    .whenMatchedDelete(condition="s.op = 'DELETE'") \
    .whenMatchedUpdateAll(condition="s.op = 'UPDATE'") \
    .whenNotMatchedInsertAll(condition="s.op = 'INSERT'").execute()
```

### Streaming
Use when low latency required from event streams.

```python
spark.readStream.format("kafka").option("subscribe", topic).load() \
    .transform(process) \
    .writeStream.format("delta") \
    .option("checkpointLocation", ckpt).toTable(target)
```

## Late Data Handling

- **Batch**: reprocess a rolling window (e.g., last 3 days) to catch late arrivals
- **Streaming**: use `.withWatermark("event_time", "1 hour")` to define late-data tolerance

## Orchestration

- Design DAGs with explicit data dependencies between tasks
- Retry with exponential backoff: 3 retries, 5min initial, 1hr max
- Support backfill via parameterized `execution_date`

## Error Handling

- **Fail fast** — validate critical assumptions (nulls, empty input) before transforms
- **Dead letter queue** — route invalid records to a DLQ table for investigation
- Log record counts at each stage; emit metrics for monitoring

## Anti-Patterns

- Blind append without partition clearing (duplicates on re-run)
- Hidden dependencies via side effects instead of explicit table reads
- Hardcoded table names instead of parameterized source/target