# Example: Avro schema validation with Confluent Schema Registry
from confluent_kafka import Consumer, Producer
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroDeserializer

schema_registry = SchemaRegistryClient({"url": "http://sr.internal:8081"})
deserializer = AvroDeserializer(schema_registry, {"auto.register.schemas": False})

consumer = Consumer({
    "bootstrap.servers": "kafka.internal:9092",
    "group.id": "pipeline-ingestion-v1",
    "auto.offset.reset": "earliest",
    "value.deserializer": deserializer
})

consumer.subscribe(["events.raw"])
# Validation happens implicitly; incompatible schemas trigger deserialization errors
# and route to a dead-letter topic for manual inspection

# Delta Lake incremental read & write pattern
from delta import configure_spark, DeltaTable
from pyspark.sql import SparkSession

spark = configure_spark_with_delta_pip(
    SparkSession.builder.appName("pipeline-transform")
).getOrCreate()

# Read only new partitions (incremental)
df = spark.read.format("delta") \
    .option("readChangeFeed", "true") \
    .option("startingVersion", last_checkpoint_version) \
    .load("s3://warehouse/raw/events")

# Apply transformations with idempotent logic
transformed = df \
    .withColumn("processed_at", current_timestamp()) \
    .withColumn("partition_date", date_format("event_ts", "yyyy-MM-dd")) \
    .dr

opDuplicates(["event_id"])

Write with partitioning and merge semantics

transformed.write
.format("delta")
.mode("append")
.partitionBy("partition_date")
.save("s3://warehouse/curated/events")

### Step 3: Implement Idempotent Transformations
Exactly-once semantics are expensive and often unnecessary. Idempotency is the pragmatic alternative. Design transformations to produce identical results regardless of execution count.

**Idempotency patterns:**
- Use deterministic composite keys (`event_id` + `source_system`)
- Apply `dropDuplicates()` or `MERGE INTO` with explicit conflict resolution
- Store processing checkpoints in external state (Redis, DynamoDB, or Delta metadata)
- Avoid non-deterministic functions (`rand()`, `current_timestamp()` in join conditions)

### Step 4: Orchestrate with Backfill & Replay Strategies
Orchestration should manage dependencies, not business logic. Use DAG-based schedulers that support parameterized runs and historical replay.

```python
# Prefect-style DAG with backfill capability
from prefect import flow, task
from datetime import timedelta

@task(retries=3, retry_delay_seconds=30)
def run_daily_transform(date: str):
    # Submit Spark job with date parameter
    submit_spark_job(
        app="transform_events",
        args=["--date", date],
        checkpoint_path=f"s3://checkpoints/{date}"
    )

@flow
def pipeline_flow(start_date: str, end_date: str):
    dates = generate_date_range(start_date, end_date)
    # Parallel execution with dependency gating
    for d in dates:
        run_daily_transform.submit(d)

Step 5: Embed Observability & Data Quality Gates

Monitoring cannot be an afterthought. Implement three tiers:

1. Infrastructure: CPU, memory, Kafka lag, Spark executor failures
2. Pipeline: Run duration, partition skew, checkpoint latency, SLA breach alerts
3. Data: Row count deltas, null rate thresholds, referential integrity checks, schema drift detection

# Data quality gate example (Great Expectations / custom)
def validate_quality(df, thresholds):
    metrics = {
        "row_count": df.count(),
        "null_rate_user_id": df.filter(col("user_id").isNull()).count() / df.count(),
        "schema_hash": compute_schema_hash(df.schema)
    }
    for metric, limit in thresholds.items():
        if metrics[metric] > limit:
            raise PipelineQualityError(f"{metric} exceeded threshold: {metrics[metric]}")
    return metrics

Architecture Decisions Summary

Decision	Recommendation	Rationale
Ingestion	CDC + Log streaming	Minimizes source load, captures deletes/updates
Transformation	Micro-batch Spark/Delta	Balances latency, fault tolerance, cost
Orchestration	DAG-based with parameterization	Enables backfill, replay, dependency management
Storage	Object storage + open table format	Avoids vendor lock, supports time travel/schema evolution
Quality	Contract-first + automated gates	Catches corruption before downstream impact

---

Pitfall Guide

1. Ignoring Schema Evolution
   Adding columns without backward compatibility breaks consumers. Always use schema registries, enforce versioning, and implement migration strategies (e.g., mergeSchema=true in Delta with explicit column mapping).

2. Stateful Transformations Without Checkpointing
   Streaming joins, window aggregations, and deduplication require checkpointing to disk. Without it, failures force full recomputation, violating SLAs and spiking costs.

3. Over-Orchestrating Business Logic
   DAGs should manage execution flow, not data transformation rules. Embedding complex joins, conditional branching, or data routing in orchestration layers creates unmaintainable graphs. Keep DAGs thin; push logic to compute engines.

4. Neglecting Idempotency & Exactly-Once Semantics
   Exactly-once is a distributed systems myth in practice. Idempotency is the engineering standard. Design writes to be replay-safe, use deterministic keys, and validate downstream state before committing.

5. Monolithic Pipeline Design
   Single DAGs handling ingestion, transformation, and delivery create blast radius explosion. Decompose into domain-scoped pipelines (raw → curated → serving) with explicit contracts between stages.

6. Treating Data Quality as Post-Processing
   Quality checks applied after delivery are too late. Implement gates at ingestion, post-transformation, and pre-serving. Fail fast, route to dead-letter queues, and alert stakeholders before corruption propagates.

7. Skipping Backfill & Replay Strategies
   Production pipelines will require historical reprocessing. If your architecture cannot replay data from arbitrary offsets or partition ranges, you cannot fix bugs, comply with regulations, or support model retraining.

---

Production Bundle

Action Checklist

• Define explicit data contracts with schema versioning and backward compatibility rules
• Implement idempotent write patterns with deterministic composite keys
• Decouple storage and compute using open table formats (Delta/Iceberg)
• Configure dead-letter queues and schema drift detection at ingestion boundary
• Embed automated data quality gates at raw, curated, and serving layers
• Establish checkpointing and replay strategies for all stateful transformations
• Track pipeline cost per terabyte and set budget alerts for compute anomalies
• Document lineage and SLA expectations for all downstream consumers

Decision Matrix

Component	Option A	Option B	Option C	Selection Criteria
Ingestion	Batch APIs	CDC (Debezium)	Event streaming (Kafka)	Source mutability, latency requirements, source system load tolerance
Transformation	Spark SQL	dbt	Flink	Complexity, team SQL proficiency, statefulness requirements
Orchestration	Airflow	Prefect	Dagster	DAG complexity, backfill needs, team DevOps maturity
Storage Format	Parquet	Delta Lake	Apache Iceberg	Schema evolution, time travel, multi-engine access, cloud provider
Quality Framework	Great Expectations	Soda Core	Custom PySpark	Team automation maturity, integration needs, alerting complexity

Configuration Template

# pipeline-config.yaml
pipeline:
  name: events-curated
  version: "2.1.0"
  owner: data-engineering@company.com
  sla:
    freshness: "15m"
    availability: "99.9%"
    
ingestion:
  source: kafka
  topics: ["events.raw.v1", "events.raw.v2"]
  consumer_group: "pipeline-ingestion-v2"
  schema_registry: "http://sr.internal:8081"
  dead_letter_topic: "events.dlq"
  
transformation:
  engine: spark
  mode: micro_batch
  batch_interval: "5m"
  idempotency_key: ["event_id", "source_system"]
  checkpoint_path: "s3://checkpoints/events-curated/"
  partitions: ["partition_date", "region"]
  
quality:
  gates:
    - metric: row_count_delta
      threshold: 0.15
      action: halt_and_alert
    - metric: null_rate_user_id
      threshold: 0.02
      action: quarantine
    - metric: schema_hash
      action: registry_validation
      
storage:
  format: delta
  path: "s3://warehouse/curated/events/"
  time_travel: true
  optimize: true
  partition_evolution: true
  
observability:
  metrics:
    - kafka_lag
    - spark_executor_failure_rate
    - pipeline_duration
    - data_quality_score
  alerts:
    slack: "#data-pipeline-alerts"
    pagerduty: true

Quick Start Guide

1. Initialize schema registry & contracts: Register Avro/Protobuf schemas for all source events. Enforce auto.register.schemas=false in consumers to prevent drift.
2. Deploy ingestion layer: Configure Kafka consumers with dead-letter routing and schema validation. Verify offset commit behavior and consumer group rebalancing.
3. Build transformation DAG: Implement micro-batch Spark jobs with incremental reads, idempotent writes, and partition-by-date strategy. Test backfill with historical data slices.
4. Embed quality gates: Add row count, null rate, and schema hash validation before committing to curated storage. Route failures to DLQ and trigger alerts.
5. Enable observability & backfill: Instrument pipeline duration, Kafka lag, and executor metrics. Verify replay capability by reprocessing a 24-hour window with modified logic.

---

Data pipeline architecture is not about choosing the fastest tool or the newest framework. It is about designing systems that fail predictably, recover deterministically, and scale without architectural collapse. Treat pipelines as production software: version them, test them, monitor them, and evolve them with explicit contracts. The platforms that survive the next decade of data complexity will be built on this discipline, not on hope.