harsh040506/data-pipeline-engineering

Data Pipeline Engineering

Design, build, and operate reliable data pipelines for ML and analytics workloads.

---

Pipeline Architecture Patterns

Batch vs. Streaming — Choosing the Right Pattern

| Concern | Batch | Streaming | |---------|-------|-----------| | Latency requirement | Hours to days | Seconds to minutes | | Data volume | Any | Very high throughput | | Complexity | Lower | Higher | | Cost | Lower | Higher | | Use cases | Daily ML training, reports | Fraud detection, recommendations |

Start with batch. Only move to streaming if latency requirements genuinely demand it.

Medallion Architecture (Data Lake)

Raw data (landing) → Bronze (raw ingest) → Silver (cleaned/validated) → Gold (aggregated/features)

| Layer | Contents | Transformation | |-------|---------|---------------| | Bronze | Raw data, unchanged. Schema-on-read. | Ingest only: add metadata (source, ingest timestamp) | | Silver | Typed, deduplicated, validated. | Apply schema, cast types, remove invalid rows | | Gold | Aggregated, business-friendly. | Joins, aggregations, feature computation |

Store all layers as Parquet (or Delta/Iceberg for ACID transactions). Always preserve Bronze — it's your source of truth for reprocessing.

---

Data Quality and Validation

Never trust data. Validate at every layer boundary.

Great Expectations (Python)

import great_expectations as gx

context = gx.get_context()
datasource = context.sources.add_pandas("my_source")
asset = datasource.add_dataframe_asset("orders")
batch_request = asset.build_batch_request(dataframe=df)

# Define expectations
suite = context.add_expectation_suite("orders_suite")
validator = context.get_validator(
    batch_request=batch_request,
    expectation_suite_name="orders_suite",
)

# Column existence
validator.expect_column_to_exist("order_id")
validator.expect_column_to_exist("customer_id")
validator.expect_column_to_exist("amount")

# No nulls in critical columns
validator.expect_column_values_to_not_be_null("order_id")
validator.expect_column_values_to_not_be_null("customer_id")

# Range validation
validator.expect_column_values_to_be_between("amount", min_value=0, max_value=100_000)

# Uniqueness
validator.expect_column_values_to_be_unique("order_id")

# Enum validation
validator.expect_column_values_to_be_in_set(
    "status", ["pending", "completed", "cancelled", "refunded"]
)

# Date format
validator.expect_column_values_to_match_strftime_format("created_at", "%Y-%m-%dT%H:%M:%S")

# Save and validate
validator.save_expectation_suite()
results = validator.validate()

if not results.success:
    failed_expectations = [r for r in results.results if not r.success]
    raise ValueError(f"Data quality check failed: {len(failed_expectations)} expectations failed")

Inline Validation (simpler)

import pandas as pd
from dataclasses import dataclass

@dataclass
class DataQualityResult:
    passed: bool
    issues: list[str]

def validate_orders(df: pd.DataFrame) -> DataQualityResult:
    issues = []
    
    # Required columns
    required = ["order_id", "customer_id", "amount", "created_at"]
    missing = [col for col in required if col not in df.columns]
    if missing:
        issues.append(f"Missing columns: {missing}")
    
    # Null checks
    null_counts = df[required].isnull().sum()
    for col, count in null_counts.items():
        if count > 0:
            issues.append(f"NULL values in '{col}': {count} rows ({count/len(df):.1%})")
    
    # Range checks
    if "amount" in df.columns:
        invalid = df["amount"] < 0
        if invalid.any():
            issues.append(f"Negative amounts: {invalid.sum()} rows")
    
    # Duplicate primary keys
    if "order_id" in df.columns:
        dups = df["order_id"].duplicated()
        if dups.any():
            issues.append(f"Duplicate order_ids: {dups.sum()}")
    
    return DataQualityResult(passed=len(issues) == 0, issues=issues)

---

Pipeline Orchestration with Airflow

Production DAG Template

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.empty import EmptyOperator
from airflow.utils.trigger_rule import TriggerRule

default_args = {
    "owner": "data-engineering",
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
    "retry_exponential_backoff": True,
    "email_on_failure": True,
    "email": ["[email protected]"],
}

with DAG(
    dag_id="ml_feature_pipeline",
    description="Daily feature engineering pipeline for churn model",
    schedule="0 4 * * *",              # 4 AM UTC daily
    start_date=datetime(2026, 1, 1),
    catchup=False,                      # Don't backfill historical runs
    max_active_runs=1,                  # Only one run at a time
    default_args=default_args,
    tags=["ml", "features", "churn"],
) as dag:

    start = EmptyOperator(task_id="start")
    
    extract_orders = PythonOperator(
        task_id="extract_orders",
        python_callable=extract_orders_from_postgres,
        op_kwargs={"date": "{{ ds }}"},       # Airflow macro for execution date
        pool="postgres_pool",                  # Limit concurrent DB connections
    )
    
    extract_events = PythonOperator(
        task_id="extract_events",
        python_callable=extract_events_from_bigquery,
        op_kwargs={"date": "{{ ds }}"},
    )
    
    validate_data = PythonOperator(
        task_id="validate_data",
        python_callable=run_data_quality_checks,
        op_kwargs={"date": "{{ ds }}"},
    )
    
    compute_features = PythonOperator(
        task_id="compute_features",
        python_callable=compute_churn_features,
        op_kwargs={"date": "{{ ds }}"},
    )
    
    load_feature_store = PythonOperator(
        task_id="load_feature_store",
        python_callable=write_to_feature_store,
        op_kwargs={"date": "{{ ds }}"},
    )
    
    notify_success = PythonOperator(
        task_id="notify_success",
        python_callable=send_slack_notification,
        op_kwargs={"status": "success"},
        trigger_rule=TriggerRule.ALL_SUCCESS,
    )
    
    notify_failure = PythonOperator(
        task_id="notify_failure",
        python_callable=send_slack_notification,
        op_kwargs={"status": "failure"},
        trigger_rule=TriggerRule.ONE_FAILED,
    )

    # DAG dependencies
    start >> [extract_orders, extract_events] >> validate_data
    validate_data >> compute_features >> load_feature_store
    load_feature_store >> [notify_success, notify_failure]

Idempotency

All pipeline tasks must be idempotent — running the same task twice with the same inputs produces the same result.

def load_to_warehouse(date: str, df: pd.DataFrame):
    """Idempotent load: delete existing data for the date, then insert."""
    with engine.begin() as conn:
        # Delete existing data for this date partition
        conn.execute(
            text("DELETE FROM features WHERE date = :date"),
            {"date": date}
        )
        # Insert fresh data
        df.to_sql("features", conn, if_exists="append", index=False)

---

Feature Engineering for ML

Feature Types and Encoding

| Feature type | Raw form | Encoding | |-------------|---------|---------| | Categorical (low cardinality) | "gold", "silver", "bronze" | One-hot encoding | | Categorical (high cardinality) | user_id, product_id | Embedding or target encoding | | Numeric | age, price | Standard scaling or quantile normalization | | Text | review text, description | TF-IDF, embeddings | | Datetime | 2026-03-01 09:14:00 | Hour, day-of-week, is_weekend, recency | | Geospatial | lat/lon | Geohash buckets, distance from reference point |

Preventing Feature Leakage

The most common ML engineering bug. Target leakage happens when features contain information from the future.

# WRONG — uses order total to predict churn, but total isn't known at prediction time
features["total_spend_this_month"] = df.groupby("customer_id")["amount"].transform("sum")

# CORRECT — use only data available at decision time (last month's spend)
features["total_spend_last_month"] = df[df["order_date"] < cutoff_date].groupby("customer_id")["amount"].sum()

Rule: For churn/conversion/retention models, features must only use data from before the label period starts.

Feature Store Pattern

from datetime import datetime
import pandas as pd

class FeatureStore:
    """Simple feature store interface."""
    
    def write_features(self, entity: str, features: pd.DataFrame, 
                       as_of_date: datetime):
        """Write features for an entity at a specific point in time."""
        features["entity"] = entity
        features["as_of_date"] = as_of_date
        features.to_parquet(f"features/{entity}/{as_of_date.date()}.parquet")
    
    def get_features(self, entity: str, feature_names: list[str],
                     as_of_date: datetime) -> pd.DataFrame:
        """Get historical feature values (point-in-time correct)."""
        # Find the most recent feature snapshot before as_of_date
        # This is the core operation that prevents leakage in training
        ...

---

Streaming Pipelines with Kafka

When you genuinely need real-time (< 1 minute latency):

from confluent_kafka import Consumer, Producer
import json

# Consumer
consumer = Consumer({
    "bootstrap.servers": "kafka:9092",
    "group.id": "ml-feature-pipeline",
    "auto.offset.reset": "latest",
    "enable.auto.commit": False,    # Manual commit — ensure at-least-once processing
})
consumer.subscribe(["orders"])

# Processing loop
while True:
    msg = consumer.poll(timeout=1.0)
    if msg is None:
        continue
    if msg.error():
        handle_error(msg.error())
        continue
    
    order = json.loads(msg.value())
    
    try:
        features = compute_realtime_features(order)
        write_to_feature_store(features)
        consumer.commit(msg)        # Commit only after successful processing
    except Exception as e:
        logger.error(f"Failed to process message: {e}", exc_info=True)
        # Send to dead letter queue, don't commit
        send_to_dlq(msg)

Streaming anti-patterns to avoid:

• Processing in the main thread without backpressure control
• Auto-commit offsets (data loss on failure)
• No dead letter queue for failed messages
• No exactly-once or at-least-once guarantee documentation

Deeper Reference

For production-ready pipeline templates and streaming infrastructure recipes, see:

• references/pipeline-patterns.md — complete Airflow DAG patterns, Kafka consumer group configurations, and Spark/dbt pipeline templates with error handling and observability