Data Pipeline Architecture That Scales Without Rewriting Everything

Design data pipelines that survive growing data volumes, changing schemas, and the inevitable 3 AM failure. Covers batch vs streaming, orchestration, schema evolution, data quality gates, and the patterns that prevent 'Big Rewrite 2.0.'

Data pipelines that scale without rewriting everything are essential when a single ETL job, once sufficient for daily batch processing, must handle real-time ingestion, multiple data sources, and complex transformations across teams and services. This architecture enables teams to evolve data processing incrementally, adding new stages, upgrading components, and adapting to changing requirements without disrupting existing pipelines or requiring a complete overhaul. It matters when data volume grows 10x, latency requirements tighten, and new analytics teams need access to raw and derived data.

Design Principles for Incremental Scalability

A scalable data pipeline architecture is built on five core principles:

  1. Decoupled components: Each stage operates independently, communicating via well-defined interfaces.
  2. Idempotent processing: A job can be re-run with the same inputs and produce the same outputs without side effects.
  3. Versioned data contracts: The schema and semantics of data at each stage are versioned and validated.
  4. Configurable and composable stages: Each component is parameterized and can be reused across pipelines.
  5. Observability at every step: Logs, metrics, and traces are available for each stage and data record.

These principles ensure that adding a new source, changing a transformation, or scaling a downstream consumer can be done in isolation.

Ingestion: From Raw to Enriched

Raw data arrives from sources such as Kafka, S3, PostgreSQL, or CSV files. The ingestion stage must capture data with minimal latency and high reliability.

Each incoming record is assigned a unique, immutable record_id derived from a hash of the source key and timestamp. This record_id is used throughout the pipeline to track data lineage.

Ingestion is configured with a pipeline.yaml file:

ingestion:
  source:
    type: kafka
    topic: user_events
    bootstrap_servers: "kafka1.example.com:9092,kafka2.example.com:9092"
    group_id: ingestion_group
    start_offset: "latest"
  format:
    type: json
    schema_version: 1.2
    required_fields:
      - event_type
      - user_id
      - timestamp
  output:
    sink: s3
    bucket: raw-data
    prefix: events/
    partitioning: "date=YYYY/MM/DD/hh"
    compression: "snappy"

When a Kafka message fails to be processed, the pipeline emits a failed_ingestion event with the following structure:

{
  "record_id": "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
  "source": "kafka",
  "topic": "user_events",
  "partition": 4,
  "offset": 12345,
  "error": {
    "type": "schema_mismatch",
    "message": "Missing required field 'user_id'",
    "expected": ["event_type", "user_id", "timestamp"],
    "actual": ["event_type", "timestamp", "session_id"]
  },
  "timestamp": "2024-03-10T14:23:07.123Z"
}

Failure is handled by a dead-letter queue (DLQ) in Kafka, where failed records are reprocessed later.

Transformation: Modular and Reusable

Transformations are defined as standalone jobs, each running in its own container. A transformation job consumes data from an input queue, applies a series of operations, and writes results to an output topic.

A transformation job is described by a transform.yaml configuration:

transform:
  input:
    type: kafka
    topic: raw_events
    group_id: transform_group
  output:
    type: kafka
    topic: enriched_events
  steps:
    - name: enrich_user_profile
      type: python
      image: "registry.example.com/transformers:1.4"
      entrypoint: "enrich.py"
      config:
        user_api_url: "https://api.example.com/users"
        cache_ttl: 3600
    - name: deduplicate
      type: sql
      query: |
        SELECT
          user_id,
          event_type,
          MAX(timestamp) as latest_timestamp,
          COUNT(*) as event_count
        FROM input
        GROUP BY user_id, event_type
        HAVING COUNT(*) > 1
      output_schema: "user_deduped_v1"
  checkpointing:
    type: "kafka"
    topic: "transform_checkpoints"
    group_id: "checkpoint_group"

Each step is idempotent: if the job is restarted, it reprocesses only uncommitted records. The checkpointing mechanism ensures that progress is tracked via Kafka offsets, and a checkpoint message is written after each batch.

When a transformation step fails, the job logs an error with a full stack trace and the input and output data of the failed batch:

ERROR [transform:enrich_user_profile] - Failed to enrich user profile for user_id=1003
Traceback (most recent call last):
  File "enrich.py", line 42, in enrich
    response = requests.get(url, timeout=5)
requests.exceptions.Timeout: HTTPSConnectionPool(host='api.example.com', port=443): Read timed out (read timeout=5)

The transform.yaml is versioned and stored in Git, enabling reproducible deployments. A new version of a transformation is deployed with docker-compose up --no-deps -d transform-service.

Data Contract Management

Each stage in the pipeline defines a data contract, stored as a JSON schema in the schemas/ directory.

For example, schemas/enriched_events_v2.json:

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "title": "Enriched Event",
  "type": "object",
  "properties": {
    "event_id": { "type": "string", "format": "uuid" },
    "user_id": { "type": "integer" },
    "event_type": { "type": "string", "enum": ["click", "purchase", "view"] },
    "user_profile": {
      "type": "object",
      "properties": {
        "name": { "type": "string" },
        "city": { "type": "string" },
        "is_premium": { "type": "boolean" }
      },
      "required": ["name", "city"]
    },
    "timestamp": { "type": "string", "format": "date-time" }
  },
  "required": ["event_id", "user_id", "event_type", "timestamp"]
}

The pipeline validates every incoming message against the expected schema. If validation fails, the record is sent to a schema_mismatch topic with the error details.

Validation is triggered by a validate.sh script:

#!/bin/bash
set -euo pipefail

SCHEMA_DIR="schemas"
INPUT_TOPIC="raw_events"
OUTPUT_TOPIC="validated_events"
LOG_FILE="/var/log/validator.log"

while IFS= read -r -d '' message; do
  echo "Validating: $(echo "$message" | jq -r '.event_id')" >> "$LOG_FILE"
  if ! echo "$message" | jq -e --arg schema "$SCHEMA_DIR/enriched_events_v2.json" -f "$SCHEMA_DIR/validate.jq"; then
    kafka-console-producer.sh \
      --bootstrap-server kafka1.example.com:9092 \
      --topic schema_mismatch \
      --property "parse.key=true" \
      --property "key.separator=|"
      <<< "{\"event_id\": $(echo "$message" | jq -r '.event_id'), \"error\": \"Schema validation failed\"}"
  fi
done < <(kafka-console-consumer.sh --bootstrap-server kafka1.example.com:9092 --topic "$INPUT_TOPIC" --property "print.key=true" --property "key.separator=|")

The validate.jq file uses jq to validate a message against the schema:

{
  event_id: .event_id,
  validation_result: (
    . as $event |
    $schema |
    .properties | to_entries |
    map({
      field: .key,
      required: .value.required | (if . then true else false end),
      type: .value.type,
      value: $event[.key],
      valid: $event[.key] | type == .value.type
    }) |
    map(select(.valid == false)) |
    length > 0
  )
}

Orchestration: Coordinating the Flow

Orchestration coordinates the entire pipeline using a workflow engine that triggers stages based on events or schedules.

A workflow is defined in workflows/etl_pipeline.yaml:

name: etl_pipeline
description: "End-to-end pipeline from raw events to analytics-ready data"
stages:
  - name: ingest
    type: job
    command: "ingest --config ingestion.yaml"
    dependencies: []
    triggers:
      - event: "new_kafka_message"
        topic: "user_events"
        conditions:
          - "message.size > 1024"
          - "message.timestamp > now - 30m"
  - name: enrich
    type: job
    command: "transform --config transform.yaml"
    dependencies:
      - "ingest"
    triggers:
      - event: "data_written"
        topic: "raw_events"
        condition: "batch_size > 1000"
  - name: analytics
    type: job
    command: "analyze --config analysis.yaml"
    dependencies:
      - "enrich"
    triggers:
      - schedule: "0 2 * * *"  # Daily at 2 AM

The orchestration engine watches Kafka topics and executes jobs when conditions are met. Jobs are run via systemctl or docker-compose, and their output is logged to /var/log/workflow/.

When a stage fails, the orchestrator sends an alert via HTTP to https://alerting.example.com/api/v1/alert with a JSON payload:

{
  "pipeline": "etl_pipeline",
  "stage": "enrich",
  "status": "failed",
  "error": {
    "message": "Transformation job failed due to timeout",
    "details": [
      "Failed to fetch user data from API",
      "Timeout after 5 seconds",
      "Last 10 records processed: [1234, 1235, ...]"
    ],
    "timestamp": "2024-03-10T15:12:03.456Z"
  },
  "metadata": {
    "run_id": "run_20240310_151203",
    "input_batch": 12345,
    "output_batch": 12345
  }
}

Monitoring and Failure Modes

Each pipeline stage emits structured logs in JSON format, with fields like stage, run_id, batch_id, and duration. Logs are collected by a central logging system (e.g., Elasticsearch) and queried via curl.

To check for failed ingestion jobs:

curl -X GET "http://logs.example.com/api/v1/search?index=ingestion&query=error.type:failed_ingestion&from=now-1h"

A common failure is data drift: the schema evolves, but the transformation pipeline does not adapt. A new field user_preferences is added to the raw event, but the enrich_user_profile job does not process it. The pipeline detects this via a schema_drift alert:

{
  "type": "schema_drift",
  "expected_fields": ["event_type", "user_id", "timestamp", "user_preferences"],
  "actual_fields": ["event_type", "user_id", "timestamp"],
  "source": "raw_events",
  "detected_at": "2024-03-10T14:00:00Z",
  "recommended_action": "Update transform.yaml to include user_preferences in input"
}

Another silent failure occurs when checkpointing is not durable. If the Kafka cluster goes down during a transformation job, and the checkpoint offset is not committed, the job restarts from an incorrect point, leading to duplicate or missed records. This is detected by comparing checkpointed_offset with latest_offset in a monitoring dashboard.

Configuration and Deployment

Configuration is managed in version-controlled directories. Each job has a config/ directory with *.yaml files. The pipeline.yaml file is used to deploy the entire system:


for stage in ingest transform analytics; do
  docker-compose up -d --build "$stage"
done

Rollbacks are done by updating the pipeline.yaml and restarting the orchestration engine. A new version of a transformation is deployed by pushing a new transform.yaml to Git and triggering a deployment script.

The pipeline is monitored via systemctl status pipeline-orchestrator and journalctl -u pipeline-orchestrator -f for real-time updates.

Summary

Scaling a data pipeline without rewriting everything relies on decoupling, idempotency, versioned contracts, and observable orchestration. Each component can evolve independently. When a new requirement arises—such as a new event type, a change in data format, or a need for real-time analytics—the pipeline adapts without requiring a full rewrite. The key is consistency: every job, every transformation, every ingestion step is defined, validated, and monitored. The architecture holds together under increasing load, evolving schemas, and team growth—without a single, monolithic rewrite.

This page was rewritten on 17 October 2026. It replaced a templated version that was not specific to its own title. How these pages are written and checked. If anything here is wrong, tell us at [email protected] and we will correct it.