Data Pipeline Architecture: ETL, ELT, and Streaming Patterns

Introduction

Data pipelines are the circulatory system of modern organizations. Every click, transaction, and sensor reading generates data that must be collected, transformed, and delivered to analysts, ML models, and business applications. Building robust, scalable data pipelines is a fundamental skill for any data engineer or software engineer working with data.

This guide covers the complete spectrum of data pipeline architectures—from traditional batch ETL to modern streaming systems. You’ll learn when to use each approach, how to handle common challenges like exactly-once processing, and practical implementation patterns using industry-standard tools.

The structure of the guide mirrors the way a practitioner actually thinks about the problem. It begins with the fundamentals — what a pipeline is and the two axes along which you choose a design: batch versus streaming, and the order in which transformation happens. From there it walks through each major architecture in turn: classic ETL with Python and Airflow, modern ELT with dbt, and continuous processing with Kafka and stream-processing engines. The later sections cover the concerns that turn a working pipeline into a production pipeline — data quality, monitoring, orchestration, and testing — because these are the activities that dominate real engineering time.

A recurring theme throughout is that architecture choices are trade-offs, not absolutes. ETL gives you a controlled transformation boundary but pays in warehouse cost and rigidity; ELT gives you flexibility at the cost of trusting your warehouse’s compute; streaming gives you low latency at the cost of significantly more operational complexity. By the end of this guide you should be able to look at any business requirement — nightly reporting, real-time alerting, a data lake feeding machine learning — and map it to a concrete architecture with the right tools and the right failure-handling strategy.

Understanding Data Pipeline Fundamentals

What is a Data Pipeline?

A data pipeline is a series of data processing steps that move data from source to destination:

Source → Extract → Transform → Load → Destination

Before diving into implementations, it is worth agreeing on the terminology used throughout this guide. The source is any system that produces data — a transactional database, an application event stream, a SaaS API, or an operational log. The destination is where the data is ultimately consumed, typically a data warehouse, a data lake, a search index, or an analytics store. In between, the pipeline applies the transformations needed to turn raw, often messy source data into something analysts and machine learning models can trust.

Two concepts recur constantly in pipeline design and deserve early attention. The first is watermarking: every pipeline needs to track how much of the source it has already processed, usually via a timestamp or sequence column, so that reruns and incremental loads stay correct. The second is idempotency: running the same load twice must yield the same result, which means writes need unique keys and upsert semantics rather than blind inserts. Nearly every best practice in this guide — schema validation, dead-letter queues, incremental loading, data contracts — exists to keep these two properties intact as pipelines grow.

Each stage may involve validation, enrichment, aggregation, or format conversion. The key is understanding that pipelines must be reliable, scalable, and maintainable.

Batch vs Streaming Processing

ETL: Extract, Transform, Load

Traditional ETL Architecture

ETL is the classic approach—extract data from sources, transform it in a processing cluster, then load into the destination:

The first code example implements a complete order ETL pipeline in plain Python, and it is deliberately simple so you can see the pattern clearly before orchestration tools are introduced. The pipeline is split into three functions that mirror the ETL phases: extract_orders reads new orders from the source database using a last_run watermark, transform_orders cleans the data, joins it with customer information, and derives metrics, and load_orders writes the enriched rows into the warehouse.

Three design decisions in this snippet are worth calling out because they carry over to every ETL you will ever write. The extraction uses an incremental query (WHERE created_at >= last_run) rather than a full table scan, which keeps runtime bounded as data grows. The transformation happens in memory with pandas before anything touches the warehouse, isolating heavy compute from the database. And the load uses an upsert — ON CONFLICT (order_id) DO UPDATE — so re-running the pipeline never creates duplicate rows. Note also that the connection credentials and the watermark state are handled ad hoc here; production pipelines would pull both from a secret store and a state database respectively.

# Python ETL with psycopg2 and pandas
import pandas as pd
import psycopg2

def extract_orders():
    """Extract from source database."""
    conn = psycopg2.connect(
        host='source-db.internal',
        database='ecommerce',
        user='etl_user',
        password='secret'
    )
    
    query = """
        SELECT order_id, customer_id, total, created_at
        FROM orders
        WHERE created_at >= %(last_run)s
    """
    
    df = pd.read_sql(query, conn, params={'last_run': get_last_run()})
    conn.close()
    return df

def transform_orders(df):
    """Transform and enrich data."""
    # Clean data
    df = df.dropna()
    df['order_date'] = pd.to_datetime(df['created_at']).dt.date
    
    # Enrich with customer data
    customers = get_customer_data(df['customer_id'].unique())
    df = df.merge(customers, on='customer_id', how='left')
    
    # Calculate derived metrics
    df['order_month'] = df['order_date'].dt.to_period('M')
    df['profit_margin'] = (df['total'] - df['cost']) / df['total']
    
    return df

def load_orders(df):
    """Load into data warehouse."""
    conn = psycopg2.connect(
        host='warehouse.internal',
        database='analytics',
        user='etl_user',
        password='secret'
    )
    
    # Upsert pattern
    for _, row in df.iterrows():
        cursor.execute("""
            INSERT INTO fact_orders (order_id, customer_id, total, order_date)
            VALUES (%s, %s, %s, %s)
            ON CONFLICT (order_id) DO UPDATE SET
                total = EXCLUDED.total
        """, (row['order_id'], row['customer_id'], row['total'], row['order_date']))
    
    conn.commit()
    conn.close()

def run_etl():
    df = extract_orders()
    df = transform_orders(df)
    load_orders(df)
    update_last_run()

Modern ETL with Apache Airflow

This three-function pattern is fine for a single pipeline, but it breaks down at scale: who schedules it, who retries a failed step, who makes sure load runs only after transform succeeds, and how do you observe the whole thing? That is where orchestration comes in. Airflow models pipelines as DAGs (directed acyclic graphs) of tasks, with explicit dependencies, retry policies, and a web UI for monitoring and manual triggers.

The Airflow example below wraps the same three functions as tasks inside a DAG, and it illustrates the core ideas of orchestration. The DAG runs on a cron schedule at 2 AM daily, retries any failed task up to three times with backoff, and does not catch up on missed runs (catchup=False). Dependencies are declared with the >> operator, so Airflow guarantees the exact execution order you specify — extract before transform before load, with an analyze step that refreshes materialized views only after the load completes. Each task is independently retried, which is far more robust than re-running a monolithic script that fails halfway through.

# dags/etl_pipeline.py
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.postgres_operator import PostgresOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'data-engineering',
    'depends_on_past': False,
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    'etl_orders_pipeline',
    default_args=default_args,
    description='Daily order ETL pipeline',
    schedule_interval='0 2 * * *',  # 2 AM daily
    start_date=datetime(2024, 1, 1),
    catchup=False,
) as dag:

    extract = PythonOperator(
        task_id='extract_orders',
        python_callable=extract_orders,
    )

    transform = PythonOperator(
        task_id='transform_orders',
        python_callable=transform_orders,
    )

    load = PythonOperator(
        task_id='load_orders',
        python_callable=load_orders,
    )

    analyze = PostgresOperator(
        task_id='update_aggregations',
        sql='sql/refresh_materialized_views.sql',
        postgres_conn_id='warehouse',
    )

    extract >> transform >> load >> analyze

Airflow also highlights a shift that happens as teams mature: the pipeline becomes a set of small, independently observable tasks rather than one big script, and correctness becomes a property of the orchestration, not of a single program. Because each task writes its own logs, state, and success records, you can pinpoint exactly which step failed and re-run only that step. This is the operational foundation every later pattern in this guide builds on, whether you are batch-loading nightly or streaming events in real time.

ETL Best Practices

The snippets below capture the three best practices that separate reliable ETL from fragile one-off scripts, and each one addresses a specific failure mode you will eventually hit.

The first is idempotency, implemented here as a watermark-aware extraction. A full load is only used when no watermark exists; after that, every run asks the source only for rows newer than the last processed value. Combined with the upsert load from the earlier example, this means a pipeline can be re-run safely after a failure without corrupting the warehouse. The second practice is schema validation at the extract boundary: check that the expected columns exist and have the right types before any transformation, so that a source schema change fails loudly at extraction rather than silently producing nulls downstream. The third is the dead-letter queue — records that fail validation are quarantined with their error details instead of aborting the whole run, so one bad row cannot stop the pipeline for every other row.

# 1. Idempotency: Running multiple times produces same result
def extract_incremental(table, watermark_column, last_value=None):
    if last_value is None:
        # Full load
        return f"SELECT * FROM {table}"
    # Incremental load
    return f"SELECT * FROM {table} WHERE {watermark_column} > '{last_value}'"

# 2. Schema validation on extract
def validate_schema(df, expected_schema):
    for col, dtype in expected_schema.items():
        if col not in df.columns:
            raise ValueError(f"Missing column: {col}")
        if df[col].dtype != dtype:
            df[col] = df[col].astype(dtype)
    return df

# 3. Error handling with dead letter queue
def process_with_dlq(df):
    valid_records = []
    errors = []
    
    for idx, row in df.iterrows():
        try:
            validated = validate_record(row)
            valid_records.append(validated)
        except ValidationError as e:
            errors.append({
                'index': idx,
                'error': str(e),
                'data': row.to_dict()
            })
    
    # Send errors to DLQ for investigation
    if errors:
        save_to_dlq(errors)
    
    return pd.DataFrame(valid_records)

ELT: Extract, Load, Transform

These three practices work together as a defense-in-depth strategy. Schema validation catches structural drift, the dead-letter queue prevents single-record failures from taking down the whole load, and idempotency ensures that whatever went wrong, the retry leaves the warehouse in the same state it would have been in had the first attempt succeeded. Teams that skip these steps inevitably rediscover them the hard way, during a late-night page about a silently corrupted table.

ELT: Extract, Load, Transform

Why ELT?

ELT flips the traditional order—load raw data first, then transform in the data warehouse. This approach leverages the data warehouse’s computational power and provides flexibility:

The fundamental shift with ELT is when transformation happens. In ETL, a dedicated processing tier transforms data before it reaches the warehouse; in ELT, the raw data is loaded as-is and all transformation runs inside the warehouse itself. This became practical because modern warehouses — Snowflake, BigQuery, Redshift — have massive, elastic compute, which means the warehouse can handle transformations that used to require a separate Spark or EMR cluster.

The SQL below shows the payoff. Raw orders are first landed into a raw_orders table exactly as they arrive, with no premature decisions about how they will be used. Then a materialized view defines the monthly aggregation, and because the raw data is retained, that view can be re-run or re-designed at any time without re-extracting anything. This is the core flexibility argument for ELT: you can change your mind about the shape of your analytics without touching the source systems or the original load.

-- Raw data landing
CREATE TABLE raw_orders AS
SELECT * FROM external_source.orders;

-- Transform layer (can be re-run anytime)
CREATE MATERIALIZED VIEW mv_monthly_orders AS
SELECT 
    DATE_TRUNC('month', order_date) AS month,
    COUNT(*) AS order_count,
    SUM(total) AS revenue,
    COUNT(DISTINCT customer_id) AS unique_customers
FROM raw_orders
GROUP BY DATE_TRUNC('month', order_date);

There is an important nuance in the raw-landing step: “raw” does not mean “unvalidated.” Serious ELT pipelines apply lightweight checks at load time — confirm the file arrived, enforce the primary key, record a load manifest — and defer only the heavier business transformations. That split keeps the warehouse clean without sacrificing the flexibility to reinterpret the data later. The materialized view then demonstrates how transformation logic stays co-located with the data, versioned in SQL alongside everything else.

ELT with dbt

dbt (data build tool) transforms data in your warehouse using SQL:

If ELT moves transformation into the warehouse, dbt is how you keep that transformation under control. dbt treats SQL models as code: each model is a .sql file that defines a view or table, models can reference each other, and dbt resolves dependencies, builds them in order, and runs tests. The configuration file below establishes the project’s structure and, crucially, its layering strategy.

The dbt_project.yml shown here defines three layers, each mapped to a schema. Staging models land and lightly clean the raw source data, intermediate models perform joins and business logic, and mart models assemble the final, user-facing analytics tables. This layering is the conceptual heart of modern ELT: it gives you a reproducible pipeline described entirely in code, where every transformation is testable, reviewable, and rerunnable by any engineer who can read SQL. The +materialized directives tell dbt whether each model should be a view (cheap, always fresh) or a table (faster queries, rebuilt on schedule).

# dbt_project.yml
name: analytics_pipeline
version: '1.0.0'

models:
  analytics:
    +materialized: table
    
    staging:
      +schema: staging
      
    intermediate:
      +schema: intermediate
      
    marts:
      +schema: marts

With the layering in place, the individual models show how each layer contributes. The first model below is a staging model, and its job is narrow but critical: take the raw orders source and make it dependable. It casts the timestamp column to a proper type, computes an order-sequence number with a window function, and filters out anything before the data cutoff.

The staging layer is the place to normalize types, enforce naming conventions, and apply the light validation mentioned earlier. It is intentionally not where business logic lives — a staging model should be a near one-to-one, cleaned reflection of the source, so that downstream consumers have a stable foundation regardless of how the source system evolves. Keeping staging simple is what lets the intermediate and mart layers change freely without breaking the pipeline’s entry point.

# models/staging/stg_orders.sql
{{ config(materialized='view') }}

SELECT
    order_id,
    customer_id,
    total,
    status,
    created_at::TIMESTAMP AS order_timestamp,
    ROW_NUMBER() OVER (PARTITION BY customer_id ORDER BY created_at) AS order_sequence
FROM {{ source('raw', 'orders') }}
WHERE created_at >= '2024-01-01'

The intermediate model below sits one level up and does the analytical heavy lifting. It aggregates orders into per-customer metrics — total orders, lifetime value, average order value, first and last order dates — and then joins those aggregates back to the customer dimension to compute derived attributes like customer tenure. Notice how it references stg_orders and stg_customers through dbt’s ref() function rather than hard-coded table names.

That indirection is deliberate and important. ref() makes dbt build models in dependency order automatically, guarantees that tests and freshness checks run against the right upstream model, and enables environments to reuse the same logic across dev and production. The intermediate layer is where business rules live — what counts as lifetime value, how tenure is computed — and keeping those rules here means the mart layer stays thin and the logic is defined exactly once rather than copy-pasted into dozens of dashboards.

# models/intermediate/int_order_metrics.sql
{{ config(materialized='table') }}

WITH customer_orders AS (
    SELECT 
        customer_id,
        COUNT(*) AS total_orders,
        SUM(total) AS lifetime_value,
        AVG(total) AS avg_order_value,
        MIN(created_at) AS first_order_date,
        MAX(created_at) AS last_order_date
    FROM {{ ref('stg_orders') }}
    GROUP BY customer_id
)

SELECT 
    c.*,
    co.total_orders,
    co.lifetime_value,
    co.avg_order_value,
    co.first_order_date,
    co.last_order_date,
    DATE_DIFF('day', co.first_order_date, co.last_order_date) AS customer_tenure_days
FROM {{ ref('stg_customers') }} c
LEFT JOIN customer_orders co ON c.customer_id = co.customer_id

The final model in this layer is the mart — the table that business users and dashboards actually query. fact_orders is a wide, denormalized fact table that joins orders with customer segments, regions, and product categories, so that a single row contains everything an analyst needs for a given order. It also demonstrates two important dbt capabilities.

The first is physical design: the partition_by configuration instructs the warehouse to partition the table by order_date, which dramatically speeds up queries that filter on date ranges. The second is a window function that computes each customer’s order number, a canonical pattern for analytics — it turns a raw list of transactions into a ranked view of customer behavior. This is the payoff of the layering discipline: analysts get one table that already encodes the business logic, and they never need to reproduce fragile joins themselves.

# models/marts/fact_orders.sql
{{ config(materialized='table') }}

{{ 
    config(
        partition_by={
            "field": "order_date",
            "data_type": "date",
            "granularity": "day"
        }
    ) 
}}

SELECT
    o.order_id,
    o.customer_id,
    o.total,
    o.status,
    o.created_at AS order_timestamp,
    DATE(o.created_at) AS order_date,
    c.segment,
    c.region,
    p.product_category,
    ROW_NUMBER() OVER (PARTITION BY o.customer_id ORDER BY o.created_at) AS customer_order_number
FROM {{ ref('stg_orders') }} o
JOIN {{ ref('stg_customers') }} c ON o.customer_id = c.customer_id
LEFT JOIN {{ ref('stg_products') }} p ON o.product_id = p.product_id

Incremental ELT with dbt

Rebuilding an entire mart table every time is wasteful once it grows past a certain size, which is where incremental materialization comes in. dbt supports materialized='incremental', a mode where the model is fully built on its first run and then only new rows are inserted on subsequent runs, keeping both query time and warehouse cost bounded.

The incremental model below shows the standard pattern for telling dbt exactly which rows are new. It first queries the maximum order_date already present in the destination table, then selects source rows newer than that watermark, and — crucially — wraps the incremental insert in an is_incremental() guard so the first full build behaves differently from later runs. The unique_key option gives dbt a way to deduplicate on insert, so a re-run of the same batch cannot create duplicate order rows. This is idempotency applied at the ELT layer: the same technique from the batch ETL section, expressed in dbt’s model language.

# models/marts/fact_orders_incremental.sql
{{ config(materialized='incremental', unique_key='order_id') }}

{% set max_order_date = run_query("SELECT MAX(order_date) FROM " ~ this ~ "").columns[0].values()[0] %}

SELECT
    order_id,
    customer_id,
    total,
    status,
    created_at AS order_timestamp,
    DATE(created_at) AS order_date
FROM {{ source('raw', 'orders') }}
WHERE 
    created_at >= COALESCE('{{ max_order_date }}', '1900-01-01')
    {% if is_incremental() %}
    AND created_at > (SELECT MAX(order_timestamp) FROM {{ this }})
    {% endif %}

Streaming Data Pipelines

Apache Kafka Fundamentals

Kafka is the backbone of modern streaming architectures:

While batch pipelines move data on a schedule, streaming pipelines move data continuously, and Kafka is the de facto standard for that job. At its core, Kafka is a distributed, append-only log: producers publish events to topics, consumers read those events, and the log retains them so that multiple consumers can read independently and replay at will. The key architectural property is decoupling — producers and consumers never interact directly, only through the log, which lets you add consumers, replay history, and change processing logic without touching producers.

The producer example below shows how straightforward publishing is. The producer connects to the cluster’s bootstrap servers, configures serializers that turn Python objects into JSON bytes, and sends order events with a key (the customer ID) and a structured value. Keys matter: they control which partition an event lands in, and partitioning by customer ID guarantees that all of a customer’s events are processed in order by the same consumer. The flush() call at the end is important in scripts — without it, buffered messages may be lost when the process exits.

# Producer: Publishing events
from kafka import KafkaProducer
import json

producer = KafkaProducer(
    bootstrap_servers=['kafka-1:9092', 'kafka-2:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8'),
    key_serializer=lambda k: k.encode('utf-8')
)

# Publish order events
for order in orders_batch:
    producer.send(
        'orders',
        key=str(order['customer_id']),
        value={
            'event_type': 'order_created',
            'order_id': order['id'],
            'customer_id': order['customer_id'],
            'total': order['total'],
            'items': order['items'],
            'timestamp': order['created_at'].isoformat()
        }
    )

producer.flush()

The consumer below is the mirror image, and its configuration choices reveal how Kafka delivers reliable, scalable consumption. A group_id places this consumer in a consumer group: Kafka then divides the topic’s partitions among the group’s members, so adding more consumers scales out processing horizontally, and when a consumer fails, its partitions are reassigned to a survivor. The auto_offset_reset='earliest' setting means a fresh group starts from the beginning of the log, while enable_auto_commit controls whether offsets are committed automatically.

The processing loop highlights two practical considerations. First, the consumer is long-lived — the for message in consumer loop runs forever, which is the normal shape for a streaming worker and requires it to run as a managed service rather than a script. Second, each event carries an event_type that routes it to the appropriate handler, showing the event-driven style where one topic can feed multiple kinds of downstream processing. For at-least-once semantics, the consumer commits offsets after processing; for exactly-once processing, you would move to transactions or the higher-level Kafka Streams API shown next.

# Consumer: Processing events
from kafka import KafkaConsumer

consumer = KafkaConsumer(
    'orders',
    bootstrap_servers=['kafka-1:9092', 'kafka-2:9092'],
    group_id='order-processor',
    value_deserializer=lambda m: json.loads(m.decode('utf-8')),
    auto_offset_reset='earliest',
    enable_auto_commit=True
)

for message in consumer:
    event = message.value
    
    if event['event_type'] == 'order_created':
        process_new_order(event)
        
    elif event['event_type'] == 'order_cancelled':
        process_cancellation(event)

Kafka Streams for Real-Time Processing

Once your events are flowing, the next step is stateful computation on them, and Kafka Streams is Kafka’s own stream-processing library. The Java example below implements the canonical word-count, but it is really a demonstration of the two fundamental abstractions of stream processing: the KStream and the KTable.

A KStream is an unbounded sequence of events, each processed as it arrives. A KTable is a changelog-backed state — a continuously updated snapshot, like a materialized table. The pipeline reads a text-lines stream, flattens each line into words, groups by word, and then .count() returns a KTable that tracks the running total for every word. That count is backed by a local state store, which is the key trick: Kafka Streams stores aggregation state on each instance and replicates it via a changelog topic, so state survives restarts and rebalancing. The result stream is finally written to an output topic for downstream consumers.

// Java: Kafka Streams word count
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.Grouped;

StreamsBuilder builder = new StreamsBuilder();

KStream<String, String> textLines = builder.stream("text-lines");

KTable<String, Long> wordCounts = textLines
    .flatMapValues(textLine -> Arrays.asList(textLine.toLowerCase().split("\\W+")))
    .groupBy((key, word) -> word, Grouped.with(String(), String()))
    .count(Materialized.as("word-counts-store"));

wordCounts.toStream().to("word-counts-output");

Stream Processing Patterns

Real streaming pipelines rarely stay at the word-count level; they need windowing, joins, and stateful aggregation. The patterns below show how these appear in the two dominant frameworks, Spark Structured Streaming and Flink.

The first pattern is windowed aggregation. Tumbling windows divide time into fixed, non-overlapping buckets, and the code aggregates order totals into one-hour windows grouped by customer. The withWatermark call is the piece that makes this correct: it tells the engine how long to wait for late-arriving events before closing a window, trading a little extra latency for the ability to include stragglers. Without a watermark, late data would either be dropped or windows would never close.

The second pattern is the streaming join. Joining two unbounded streams requires buffering — the engine must hold records on both sides until their join keys match — so streaming joins consume memory and need careful key design. The third pattern shows Flink’s alternative: a KeyedProcessFunction with explicit state. Here an aggregator stores each customer’s running count and total in a map state, updating it per event. Understanding these three patterns — windowing with watermarks, streaming joins, and keyed state — covers the majority of real stream-processing workloads.

# Pattern 1: Windowed Aggregation
from pyspark.sql import SparkSession
from pyspark.sql.functions import window, sum, count

spark = SparkSession.builder.getOrCreate()

orders = spark.readStream.format('kafka')\
    .option('kafka.bootstrap.servers', 'kafka:9092')\
    .option('subscribe', 'orders')\
    .load()

# Tumbling window (non-overlapping)
orders_with_watermark = orders\
    .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")\
    .withWatermark("timestamp", "10 minutes")

windowed_sales = orders_with_watermark\
    .groupBy(
        window("timestamp", "1 hour"),
        "customer_id"
    )\
    .agg(
        sum("total").alias("hourly_spend"),
        count("order_id").alias("order_count")
    )

# Pattern 2: Streaming Join
orders = spark.readStream.format('kafka')\
    .option('kafka.bootstrap.servers', 'kafka:9092')\
    .option('subscribe', 'orders')\
    .load()

products = spark.readStream.format('kafka')\
    .option('kafka.bootstrap.servers', 'kafka:9092')\
    .option('subscribe', 'products')\
    .load()

enriched_orders = orders.join(products, orders.product_id == products.product_id)

# Pattern 3: State Store (Flink)
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.state import MapStateDescriptor

env = StreamExecutionEnvironment.get_execution_environment()

class OrderAggregator(MapStateDescriptor):
    def __init__(self):
        super().__init__("order-aggregator", String(), RowTypeInfo(BasicTypeInfo.LONG_TYPE, BasicTypeInfo.LONG_TYPE))

@KeyedProcessFunction
def aggregate_orders(ctx, order, aggregator: MapState):
    current = aggregator.get(order.customer_id)
    if current is None:
        current = (0, 0)
    
    new_count = current[0] + 1
    new_total = current[1] + order.total
    aggregator.put(order.customer_id, (new_count, new_total))

Data Quality and Monitoring

Data Quality and Monitoring

Data quality is not a phase you bolt on at the end; it is a property you design for, because bad data flowing through a pipeline silently corrupts every downstream report and model. This section covers the two complementary mechanisms teams use: data contracts that validate data at the boundaries, and monitoring that alerts on problems as they occur.

Data Contracts

A data contract is a formal, machine-checkable agreement about what a piece of data must look like — its columns, types, nullability, and value constraints. The OrderContract below is a Python implementation of that idea, applied to an orders dataframe. Its validate method checks four things in sequence: that every required column exists, that none of them contain nulls, that total has a numeric type, and that status only takes values from an allowed set.

Each check fails fast with a descriptive error, so a broken contract surfaces at the earliest possible point — the moment the data enters the pipeline — rather than days later when an analyst sees strange numbers. The contract is also a communication tool: it tells source-system owners exactly what the pipeline needs, turning “the pipeline broke” into a precise, testable requirement. In practice, contracts like this are the interface between teams that produce data and teams that consume it, and maintaining them is a form of API versioning for data.

# Define expected schema
from dataclasses import dataclass
from typing import List
import pyspark

@dataclass
class OrderContract:
    order_id: str
    customer_id: str
    total: float
    status: str
    created_at: str
    
    @staticmethod
    def validate(df: pyspark.sql.DataFrame) -> bool:
        required_columns = ['order_id', 'customer_id', 'total', 'status', 'created_at']
        
        for col in required_columns:
            if col not in df.columns:
                raise DataContractError(f"Missing required column: {col}")
        
        # Null checks
        null_counts = {col: df.filter(df[col].isNull()).count() for col in required_columns}
        for col, count in null_counts.items():
            if count > 0:
                raise DataContractError(f"Found {count} null values in {col}")
        
        # Type validation
        if not isinstance(df.select('total').first()[0], (int, float)):
            raise DataContractError("total must be numeric")
        
        # Value constraints
        invalid_statuses = df.filter(~df.status.isin(['pending', 'completed', 'cancelled'])).count()
        if invalid_statuses > 0:
            raise DataContractError(f"Found {invalid_statuses} invalid status values")
        
        return True

Pipeline Monitoring

Validation answers “is the data correct?”; monitoring answers “is the pipeline healthy, and if not, who gets paged?” The monitoring example below shows both halves of the answer.

The first half is operational alerting. The slack_failure_callback is wired into Airflow as an on_failure_callback, so whenever any task in a DAG fails, a message lands in the #data-alerts channel with the failing DAG, task, and error. This is the minimum viable incident response: the right people are notified with enough context to start debugging without opening the Airflow UI. The second half is data-quality alerting with Great Expectations, which encodes business expectations about the data itself — no null order IDs, total values that are always non-negative, and a realistic distribution of order statuses.

# Airflow metrics and alerting
from airflow.hooks.base import BaseHook
from slack_sdk import WebClient

def slack_failure_callback(context):
    dag_id = context['dag'].dag_id
    task_id = context['task_instance'].task_id
    error = context['exception']
    
    client = WebClient(token=os.environ['SLACK_TOKEN'])
    client.chat_postMessage(
        channel='#data-alerts',
        text=f"❌ Pipeline Failed: {dag_id}.{task_id}\nError: {str(error)}"
    )

# Data quality checks
from great_expectations import great_expectations as gx

def validate_orders_data():
    context = gx.get_context()
    batch = context.sources.pandas_files.read_csv("orders.csv")
    
    expectations = [
        expect_column_values_to_not_be_null.column("order_id"),
        expect_column_values_to_be_between.column("total").min_value(0),
        expect_column_distributions_to_match_histogram.column("status")
            .expected_histogram_partition_object([
                {"value": "pending", "count": 0.1},
                {"value": "completed", "count": 0.8},
                {"value": "cancelled", "count": 0.1}
            ])
    ]
    
    results = batch.validate(expectations=expectations)
    
    if not results["success"]:
        send_alert(f"Data quality failed: {results['statistics']}")

The two alerting styles are complementary rather than redundant. Airflow catches infrastructure and execution failures — a dead database, a timeout, a malformed file. Great Expectations catches semantic failures — the data arrived but is wrong, incomplete, or drifting from its expected distribution. Between them they cover both “the pipeline did not run” and “the pipeline ran but produced garbage,” which are the two failure modes that actually matter in production. A good monitoring setup sends the first kind to the on-call engineer and the second kind to the owning data team, with enough detail that neither has to start from zero.

Lambda and Kappa Architectures

Lambda Architecture

Combines batch and streaming layers:

The Lambda architecture is an answer to the tension between batch and streaming that emerged when the two were hard to reconcile. It runs both paths in parallel: a batch layer recomputes results from raw historical data for accuracy and completeness, while a speed layer processes events in real time for low latency. A serving layer then merges the two views into the answers applications see.

The diagram below shows the shape. The batch layer typically reads from object storage or a lake and produces accurate, expensive-to-compute results on a schedule. The speed layer handles the window between the last batch run and now, trading some accuracy for immediacy. The serving layer merges them, so users get recent data instantly and corrections arrive when the batch layer catches up. The notorious downside is that you now write every piece of logic twice — once for batch and once for streaming — and the two versions inevitably drift apart, which is the main argument against the pattern.

┌─────────────────────────────────────────┐
│              Data Sources               │
└─────────────────┬───────────────────────┘
                  │
        ┌─────────┴─────────┐
        │                   │
        ▼                   ▼
┌───────────────┐    ┌───────────────┐
│   Batch Layer │    │  Speed Layer  │
│   (HDFS/S3)   │    │  (Streaming)  │
└───────┬───────┘    └───────┬───────┘
        │                    │
        └─────────┬──────────┘
                  │
                  ▼
        ┌───────────────────┐
        │   Serving Layer   │
        │   (Query/Merge)   │
        └───────────────────┘

The merging step is also harder than it looks. The batch and speed layers compute the same metric on the same underlying events but at different times, so the serving layer must deduplicate and reconcile overlapping windows without double counting. Teams that adopt Lambda typically do so pragmatically: use streaming only where latency is a hard requirement, batch everywhere else, and keep the shared logic in a common library so the two paths at least start from the same code.

Kappa Architecture

Simplifies to single streaming path:

The Kappa architecture is the direct response to Lambda’s duplication problem: instead of running two code paths, treat the event log itself as the source of truth and make everything a stream. In Kappa, all data flows through a single streaming pipeline backed by an immutable, replayable log such as Kafka. Because the log retains history, there is no separate batch system — “batch” becomes “replay the log through the streaming pipeline,” and recomputation is just re-running the same code over historical events.

┌─────────────────────────────────────────┐
│              Data Sources               │
└─────────────────┬───────────────────────┘
                  │
                  ▼
        ┌───────────────────┐
        │  Kafka Stream     │
        │  (Immutable Log)  │
        └─────────┬─────────┬───────────┐
                   │         │           │
                   ▼         ▼           ▼
            ┌──────────┐ ┌──────────┐ ┌──────────┐
            │ Real-time │ │   Replay │ │  Query   │
            │  Service  │ │ (Rebuild)│ │  Layer   │
            └──────────┘ └──────────┘ └──────────┘

The diagram shows the payoff. The immutable log feeds real-time services, a replay path that rebuilds any state from scratch, and a query layer for historical analysis — all from the same code. There is only one pipeline to write, test, and maintain, which eliminates the Lambda drift problem entirely. The trade-offs are that the underlying stream engine must be able to keep up with full-history replays, and some workloads that are naturally batch-shaped (large full-table aggregates) remain cheaper to do as scheduled batch jobs, even in a Kappa world. In practice many teams run a Kappa-style streaming core with a small batch layer for exactly those cases.

Orchestration with Prefect

Airflow dominated orchestration for years, but a newer generation of tools — Prefect being the most popular — addresses its ergonomic weaknesses by describing pipelines as ordinary Python. Instead of a separate DAG DSL, you decorate functions with @task and @flow, and the flow function’s control flow is the dependency graph. This makes pipelines testable and debuggable with plain Python tooling, which is a significant productivity gain for data teams.

The example below shows the style. Three @task-decorated functions define the extract, transform, and load steps, and daily_pipeline — decorated with @flow — calls them in sequence. Prefect infers that transform depends on extract, and load on transform, without any explicit operators. The transaction() context in the load step is a standout feature: it makes the write and the completion marker atomic, so a crash mid-load leaves no partial state behind. Finally, daily_pipeline.serve schedules the flow on a cron schedule and exposes it to the Prefect server for monitoring and retries.

# prefect_pipeline.py
from prefect import flow, task
from prefect.transactions import transaction
import psycopg2

@task
def extract():
    return fetch_from_source()

@task
def transform(data):
    return transform_data(data)

@task
def load(data):
    with transaction():
        insert_into_warehouse(data)
        mark_pipeline_complete()
        
@flow
def daily_pipeline():
    data = extract()
    transformed = transform(data)
    load(transformed)
    
# Schedule
if __name__ == "__main__":
    daily_pipeline.serve(
        cron="0 2 * * *",
        parameters={"environment": "production"}
    )

Data Pipeline Testing

Orchestration gives you reliable execution, but it cannot tell you whether the transformations are correct. That is what testing is for, and data pipelines are eminently testable once you separate pure transformation logic from I/O. The tests below follow the standard arrange-act-assert pattern against a local Spark session.

The first test validates the core order-total calculation: it builds a small dataframe of orders with nested items, runs the transformation, and asserts the computed totals. This is the most valuable kind of pipeline test because it locks down the business logic that everything downstream depends on. The second test verifies failure handling — it feeds in a record with a missing customer ID and asserts that validation raises the expected error. Together they illustrate the rule that makes pipeline testing practical: keep transformations as pure functions that take a dataframe and return a dataframe, and you can test them with tiny fixtures instead of needing a live cluster.

# Unit tests for transformations
import pytest
from pyspark.sql import SparkSession

@pytest.fixture
def spark():
    return SparkSession.builder.master("local[*]").getOrCreate()

def test_order_total_calculation(spark):
    # Arrange
    input_df = spark.createDataFrame([
        {"order_id": "1", "items": [{"price": 100}, {"price": 50}]},
        {"order_id": "2", "items": [{"price": 75}]}
    ])
    
    # Act
    result = calculate_order_totals(input_df)
    
    # Assert
    assert result.filter(result.order_id == "1").first()["total"] == 150
    assert result.filter(result.order_id == "2").first()["total"] == 75

def test_null_handling(spark):
    input_df = spark.createDataFrame([
        {"order_id": "1", "customer_id": None}
    ])
    
    with pytest.raises(ValidationError):
        validate_required_fields(input_df)

Beyond unit tests, a complete pipeline test suite also includes integration and end-to-end layers. Integration tests run the real transformation against a small sample of actual data to catch environment-specific issues — a date parsing quirk, a driver version mismatch — that pure fixtures miss. End-to-end tests run the pipeline against ephemeral infrastructure and assert on the final tables, which is the closest thing to a guarantee that deployment will behave. A pragmatic strategy is to run unit tests on every commit, integration tests on every merge, and end-to-end tests on every release, matching test cost to how often and how early you need feedback.

Conclusion

Building data pipelines requires understanding your use case: batch ETL for complex transformations, ELT for flexibility with modern data warehouses, and streaming for real-time requirements. Tools like Airflow for orchestration, Kafka for streaming, and dbt for transformations form the foundation of modern data infrastructure.

Start simple, instrument everything, and iterate. Data quality and monitoring aren’t optional—they’re what separate production-grade pipelines from fragile experiments.