ESC
Type to search guides, tutorials, and reference documentation.

Data Pipelines

ETL and ELT, idempotency, incremental extraction and change data capture, late data, backfills, orchestration, data contracts and the quality checks that catch silent corruption.

Data pipelines are the structured, repeatable processes that move data from sources to destinations, transforming and validating it along the way. They matter when data must be available in a consistent, timely form—such as daily reports, real-time dashboards, or machine learning model training—across systems with different formats, schemas, and update frequencies. A pipeline is not just a script; it is a deployable, observable, and recoverable workflow that runs on a schedule or in response to events, with clear checkpoints, error handling, and monitoring.

Building a Pipeline from Source to Destination

Ingesting from a File System

To pull data from a local or networked file system, use rsync with a known set of options for idempotent, incremental updates:

rsync -avz --delete --exclude='*.tmp' --checksum \
      --log-file=/var/log/pipeline-ingest.log \
      /data/source/ \
      /data/ingest/

The --checksum flag ensures that files are only transferred when content changes, not just when timestamps differ. The --log-file captures every transfer, enabling auditing and alerting. A common failure is missing --delete, which leaves stale files in the destination, causing duplicate or outdated data. Another silent error occurs when the source system writes files as *.tmp and then renames them; without --exclude='*.tmp', the pipeline ingests incomplete files.

Processing with a Streaming ETL Framework

For transformation and enrichment, use Apache Spark with spark-submit to run a Python-based ETL job:

spark-submit \
  --class com.example.DataPipeline \
  --conf spark.executor.memory=4g \
  --conf spark.sql.adaptive.enabled=true \
  --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.0 \
  --files config.json \
  etl-job.py \
  --input-path /data/ingest \
  --output-path /data/processed \
  --config config.json

The --files flag places config.json on every executor’s classpath. The job reads raw data, applies schema inference, joins with a reference dataset, and writes to Parquet. A frequent issue is spark.sql.adaptive.enabled=true being misconfigured: it enables adaptive query execution but requires spark.sql.adaptive.coalescePartitions.enabled=true to coalesce small output files—otherwise, the output is fragmented into hundreds of tiny Parquet files.

Validating Data with a Schema Check

After processing, validate the data against a known schema using great-expectations in a validate.py script:

import great_expectations as gx
from great_expectations.core import ExpectationSuite

context = gx.DataContext()
suite = context.get_expectation_suite("sales_schema")
batch = context.get_batch(
    data_asset_name="sales",
    batch_kwargs={"path": "/data/processed/sales.parquet"}
)

results = suite.validate(batch)
if not results.success:
    raise RuntimeError(
        f"Validation failed: {len(results.results)} expectations failed\n"
        f"Details: {results.results}"
    )

The validation fails silently if results.success is True but one critical expectation is False. The pipeline must check results.results to log which specific expectations failed. A common mistake is relying solely on results.success without inspecting results.results, leading to undetected schema drift.

Orchestrating the Pipeline

Scheduling with Cron

Use crontab to run the pipeline daily:


0 2 * * * /opt/pipelines/run-pipeline.sh

# Run validation every 15 minutes
*/15 * * * * /opt/pipelines/run-validation.sh

The script run-pipeline.sh runs rsync, spark-submit, and validate.py in sequence. A silent failure occurs when the script exits with status 0 but the pipeline’s output is corrupt: the cron job does not capture stderr, so errors from Spark or great-expectations are lost. Always redirect stderr and stdout:

#!/bin/bash
exec 1>/var/log/pipeline.log 2>&1
echo "Starting pipeline at $(date)"

rsync -avz --delete --exclude='*.tmp' --checksum \
      /data/source/ /data/ingest/

spark-submit \
  --class com.example.DataPipeline \
  --conf spark.executor.memory=4g \
  --conf spark.sql.adaptive.enabled=true \
  --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.0 \
  --files config.json \
  etl-job.py \
  --input-path /data/ingest \
  --output-path /data/processed \
  --config config.json

python validate.py

echo "Pipeline completed at $(date)"

Monitoring with Health Checks

Expose a /health endpoint via a simple HTTP server to check pipeline status:

from http.server import HTTPServer, BaseHTTPRequestHandler
import json
import subprocess
import os

class HealthHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path == "/health":
            status = {"healthy": True, "checks": {}}
            checks = []

            # Check if /data/ingest has files
            ingest_files = len(os.listdir("/data/ingest"))
            checks.append({
                "name": "ingest_directory",
                "ok": ingest_files > 0,
                "message": f"Found {ingest_files} files in ingest"
            })

            # Check if Spark job produced output
            output_files = len(os.listdir("/data/processed"))
            checks.append({
                "name": "output_directory",
                "ok": output_files > 0,
                "message": f"Produced {output_files} output files"
            })

            # Check if validation succeeded
            validation_ok = subprocess.run(
                ["python", "validate.py", "--check-only"],
                capture_output=True,
                text=True
            ).returncode == 0
            checks.append({
                "name": "validation",
                "ok": validation_ok,
                "message": "Validation passed"
            })

            status["checks"] = checks
            status["healthy"] = all(c["ok"] for c in checks)

            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.end_headers()
            self.wfile.write(json.dumps(status).encode())
        else:
            self.send_error(404)

if __name__ == "__main__":
    server = HTTPServer(("0.0.0.0", 8080), HealthHandler)
    server.serve_forever()

This endpoint returns JSON with nested checks. The pipeline orchestrator can poll /health every minute and alert on {"healthy": false}. A subtle failure mode is a validation script that returns 0 even when it logs errors to stderr—the orchestrator sees it as healthy, but the data is invalid.

Handling Events and Errors

Responding to Event Triggers

For event-driven pipelines, use inotifywait to react to file arrivals:

inotifywait -m -r -e create,move --format '%w%f' /data/source/ \
  | while read file; do
      echo "New file: $file"
      spark-submit \
        --class com.example.DataPipeline \
        --conf spark.executor.memory=4g \
        etl-job.py \
        --input-path "$file" \
        --output-path /data/processed/$(basename "$file")
    done

This continuously monitors /data/source/ and triggers a Spark job on every new file. A common mistake is using inotifywait without --monitor (-m), causing the pipeline to exit after the first event. Another failure is not handling --format '%w%f', leading to malformed paths when files are moved into subdirectories.

Managing Failure and Recovery

When a pipeline step fails, write a log entry with structured metadata:

import logging
import traceback

logger = logging.getLogger("pipeline")

try:
    # Run ETL
    spark_job()
    validate_data()
except Exception as e:
    logger.error(
        "Pipeline failed",
        extra={
            "step": "etl",
            "timestamp": "2024-01-10T12:34:56Z",
            "error_type": type(e).__name__,
            "error_message": str(e),
            "traceback": traceback.format_exc()
        }
    )
    raise

The extra dictionary attaches structured data to the log entry. A pipeline that logs only logger.error(str(e)) misses context. The traceback is often omitted, so failures in complex functions are hard to debug.

Handling Partial Failures

When a pipeline runs over many files, it may process some but fail on others. Use --dry-run to test the pipeline:

spark-submit \
  --class com.example.DataPipeline \
  --conf spark.executor.memory=4g \
  --conf spark.sql.adaptive.enabled=true \
  etl-job.py \
  --input-path /data/ingest \
  --output-path /data/processed \
  --dry-run \
  --log-file /data/logs/dry-run.log

With --dry-run, the job runs but does not write output. This allows testing schema, transformations, and validation without affecting production data. A silent error is a dry-run that logs to stdout but does not write --log-file—the orchestrator sees a successful dry-run, but logs are lost.

Running in Production

Configuration and Deployment

Pipeline configuration is managed via a pipeline.yaml file:

source:
  path: /data/source
  polling_interval: 300
  file_pattern: "*.csv"
  exclude:
    - "*.tmp"
    - "*.bak"

transform:
  spark:
    executor_memory: "4g"
    adaptive: true
    packages:
      - org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.0
  script: etl-job.py

validate:
  schema: sales_schema.json
  threshold: 0.95
  checks:
    - column_exists
    - not_null
    - unique
    - type_match

output:
  path: /data/processed
  format: parquet
  partition_by:
    - year
    - month
    - day

The partition_by list is often confused with partition_by: ["year", "month", "day"]—the former is a list of strings, the latter is a list of values. A common failure is missing format: parquet, causing Spark to write in the default format (which may be ORC), leading to compatibility issues.

Debugging with Logs and Traces

Use journalctl to monitor the pipeline daemon:

journalctl -u pipeline-service -f --since "2024-01-10"

The -f flag follows the log, and --since filters to a specific date. A subtle failure is missing --since when investigating a recent failure, so the log contains irrelevant entries from weeks prior.

For deeper tracing, enable Spark’s spark.sql.adaptive.coalescePartitions.enabled=true and spark.sql.adaptive.skewJoin.enabled=true, which dynamically optimize query execution but require monitoring of spark.executor.logs and spark.driver.logs. A pipeline may run slowly but not be optimized; the logs reveal CoalescePartitions and SkewJoin execution plans.

Recovery from Data Drift

When a new source schema arrives, the pipeline may fail silently. Use schema inference and schema evolution in Spark:

df = spark.read \
  .option("mode", "PERMISSIVE") \
  .option("mergeSchema", "true") \
  .parquet("/data/ingest")

The PERMISSIVE mode allows missing columns and tolerates malformed data. With mergeSchema=true, Spark automatically updates the schema on each run. A silent failure occurs when mergeSchema=true but mode is FAILFAST, causing the pipeline to halt on a single malformed record, while PERMISSIVE silently logs errors.

When schema drift occurs, the pipeline may write data with a new column, but downstream consumers expect a fixed schema. Use schema validation at the output step:

df.write \
  .mode("overwrite") \
  .option("schema", schema) \
  .parquet("/data/processed")

The schema option ensures the output matches the expected schema, failing if any column is missing or mismatched. A common mistake is omitting schema and relying on mergeSchema alone—data arrives with extra columns, but consumers do not use them.

Summary

A data pipeline is a sequence of stages—ingest, process, validate, output—orchestrated across time and events. It relies on precise commands, consistent configuration, and robust error handling. The key to success is not just running the pipeline, but knowing when and how it fails. The most valuable insight is that a pipeline is not a script—it is a system: it logs, monitors, recovers, and adapts.

This page was rewritten on 18 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.