Data Pipeline Orchestration

Production engineering guide for data pipeline orchestration covering patterns, implementation strategies, and operational best practices.

Data pipeline orchestration is the coordination of multiple data processing steps that must execute in dependency order, with defined inputs, outputs, and failure handling. It matters when you need to reliably move and transform data between systems while maintaining end-to-end observability and recovery capabilities.

Creating a pipeline definition

A pipeline definition describes the sequence of tasks, their dependencies, and execution parameters. You can define this using YAML format with standard fields:

name: customer-data-ingestion
version: 1
tasks:
  - name: fetch-customer-records
    type: http
    endpoint: /api/customers
    method: GET
    output: customer_records.json
  - name: validate-customer-data
    type: python
    script: validate.py
    input: customer_records.json
    output: validated_customers.json
  - name: load-to-warehouse
    type: sql
    query: |
      INSERT INTO customers VALUES (...)
    input: validated_customers.json

The type field determines how the task executes. Valid types include http, sql, python, bash, and docker. Each task must specify either input or output (or both), and all paths are relative to a working directory.

Tasks with no dependencies execute in parallel after the pipeline starts. Tasks with dependencies wait for their specified inputs to be written before starting.

Running a pipeline

Use the command-line interface to start a pipeline:

pipeline run --definition-file my-pipeline.yaml --workspace /tmp/pipeline-workspace

The --workspace flag is required and must point to an existing directory. The system creates subdirectories for each task's working space within this location.

If a task fails, the pipeline stops and reports the failure. You can resume from where it left off with:

pipeline resume --definition-file my-pipeline.yaml --workspace /tmp/pipeline-workspace

The resume command only re-executes tasks that did not complete successfully in the previous run.

Handling task dependencies

Dependencies between tasks are declared in the pipeline definition. The system tracks which outputs each task creates and waits for required inputs before starting:

tasks:
  - name: extract-sales-data
    type: sql
    query: SELECT * FROM sales
    output: sales.csv
  - name: transform-sales
    type: python
    script: transform.py
    input: sales.csv
    output: sales_transformed.json

If a task's input file does not exist, the system logs an error:

ERROR: Task 'transform-sales' requires input 'sales.csv', but file does not exist

Dependencies can be transitive. If task-c depends on task-b, and task-b depends on task-a, then task-c waits for task-a to complete.

Managing pipeline state

The system tracks pipeline execution state using a metadata store. This is typically a SQLite database in the workspace directory named pipeline.db. The database contains tables for:

You can query the current state with:

pipeline status --workspace /tmp/pipeline-workspace

The output shows which tasks have completed, which are running, and which have failed.

Configuring failure handling

Each task can specify a maximum retry count. If a task fails and has retries remaining, it will be retried:

tasks:
  - name: load-to-warehouse
    type: sql
    query: INSERT INTO customers VALUES (?)
    input: validated_customers.json
    max_retries: 3

If all retries are exhausted, the task fails and the pipeline stops. You can also configure a failure callback:

failure_callback: |
  curl -X POST https://monitoring.example.com/pipeline-failure \
       -H "Content-Type: application/json" \
       -d '{"pipeline": "customer-data-ingestion", "timestamp": "2023-10-01T10:00:00Z"}'

Scheduling and automation

Pipelines can be scheduled using standard cron syntax:

pipeline schedule --cron "* 0 * * *" --definition-file my-pipeline.yaml

The system maintains a queue of scheduled runs. Each time the scheduler fires, it creates a new pipeline run with a timestamp in its metadata.

Scheduled pipelines do not automatically resume failed runs. If a scheduled pipeline fails, it must be manually resumed or re-run.

Monitoring and logging

Each task writes logs to a file in the workspace under logs/task-name.log. The log format is:

2023-10-01T10:00:00Z [INFO] Starting task fetch-customer-records
2023-10-01T10:00:01Z [INFO] HTTP request to /api/customers completed with status 200
2023-10-01T10:00:01Z [INFO] Task completed successfully in 1.2 seconds

Error logs contain stack traces and system messages. The system writes all logs to stderr during execution.

Recovering from failures

If a pipeline fails, you can recover by re-running specific tasks:

pipeline run --definition-file my-pipeline.yaml \
            --workspace /tmp/pipeline-workspace \
            --rebuild-tasks fetch-customer-records,validate-customer-data

The --rebuild-tasks flag forces re-execution of those tasks and all downstream tasks. It does not delete outputs from successful upstream tasks.

Task execution environments

Tasks run in isolated containers or processes. You can configure the environment for a task:

tasks:
  - name: validate-customer-data
    type: python
    script: validate.py
    environment:
      PYTHONPATH: /usr/local/lib/python3.9/site-packages
      LOG_LEVEL: DEBUG

Environment variables are set before the task starts, but they do not affect tasks that have already started.

Pipeline parameterization

You can pass parameters to a pipeline at runtime:

pipeline run --definition-file my-pipeline.yaml \
            --workspace /tmp/pipeline-workspace \
            --parameters date=2023-10-01,limit=1000

Parameters are available in tasks as environment variables. The system does not validate parameter names or types.

Version control integration

Pipeline definitions should be version-controlled. The system does not enforce versioning but works with any VCS:

git add my-pipeline.yaml
git commit -m "Update customer data pipeline"

You can check the current pipeline version by reading the version field from the definition file.

Common failure modes

The most common silent failure is when a task produces no output file. The system does not validate that output files are created:

tasks:
  - name: generate-report
    type: bash
    command: echo "No output file created"
    output: report.txt

If the script exits successfully but does not create report.txt, the pipeline continues without error.

Another silent failure occurs when dependencies are specified incorrectly. If a task references an input that never exists, the system only reports the error at runtime:

ERROR: Task 'transform-sales' requires input 'sales.csv', but file does not exist

If a dependency chain includes a task that never produces its output, all downstream tasks will wait indefinitely for that missing file.

Resource limits

You can set resource constraints on pipeline execution:

limits:
  max_concurrent_tasks: 5
  timeout_seconds: 3600

If a pipeline exceeds its timeout, it stops and reports:

ERROR: Pipeline timed out after 3600 seconds

The system does not enforce memory limits on individual tasks, only overall pipeline execution.

Debugging pipelines

For debugging, you can run a pipeline in debug mode:

pipeline run --definition-file my-pipeline.yaml \
            --workspace /tmp/pipeline-workspace \
            --debug

In debug mode, the system preserves all intermediate files and does not clean up the workspace on completion. You can also set task-specific debug flags:

tasks:
  - name: fetch-customer-records
    type: http
    endpoint: /api/customers
    method: GET
    debug: true

The debug flag enables verbose logging for that specific task and preserves its working directory.

Task output validation

You can specify output validation rules:

tasks:
  - name: validate-customer-data
    type: python
    script: validate.py
    input: customer_records.json
    output: validated_customers.json
    validation:
      schema: customer-schema.json
      min_rows: 10

If the validation fails, the task reports an error but does not automatically retry. You must handle this in your pipeline definition or failure callback.

Configuration file location

Pipeline configuration files are typically stored in version control. The system does not have a central configuration directory; all paths in the definition file are relative to the workspace directory or absolute.

The pipeline command assumes that all pipeline definitions are located in the current working directory unless specified otherwise with --definition-file.

Task types and their behavior

HTTP tasks

HTTP tasks make requests to endpoints. They require a method, endpoint, and optionally headers:

tasks:
  - name: fetch-customer-data
    type: http
    method: GET
    endpoint: /api/customers
    headers:
      Authorization: Bearer secret-token

HTTP tasks exit with error code 1 if the HTTP response status is 4xx or 5xx.

SQL tasks

SQL tasks execute queries and write results to files. They require a query field:

tasks:
  - name: extract-sales-data
    type: sql
    query: SELECT * FROM sales WHERE date = '2023-10-01'
    output: sales.csv

SQL tasks exit with error code 1 if the query fails or returns no rows.

Bash tasks

Bash tasks execute shell commands:

tasks:
  - name: compress-data
    type: bash
    command: gzip -c data.json > data.json.gz
    input: data.json
    output: data.json.gz

Bash tasks exit with the code of the last command in the pipeline.

Python tasks

Python tasks run scripts with standard Python execution:

tasks:
  - name: validate-customer-data
    type: python
    script: validate.py
    input: customer_records.json
    output: validated_customers.json

Python tasks must be valid Python files and are executed in a subprocess with the current working directory set to the task's workspace.

Docker tasks

Docker tasks run containers:

tasks:
  - name: process-data
    type: docker
    image: my-processor:latest
    command: python process.py
    input: data.json
    output: processed.json

Docker tasks must have a valid Docker image available in the local environment.

Pipeline lifecycle management

Pipelines can be stopped, paused, or resumed. The system does not support pausing individual tasks, only entire pipelines:

pipeline stop --workspace /tmp/pipeline-workspace

This command cancels all running tasks but preserves state for future resumption.

You can also delete pipeline runs from history:

pipeline cleanup --workspace /tmp/pipeline-workspace --keep-runs 5

This removes metadata for runs older than the specified number of recent runs.

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