MLOps Pipeline Engineering

Build production ML pipelines that automate training, evaluation, deployment, and monitoring. Covers feature stores, model registries, training pipelines, A/B testing for models, model drift detection, and the patterns that make ML repeatable and reliable.

This page describes the engineering practices and operational mechanics for building, deploying, and maintaining ML pipelines. It focuses on how to construct and manage pipeline workflows rather than model training or inference details.

Creating Pipeline Definitions

Pipelines are defined through configuration files written in YAML or JSON formats. A basic pipeline definition includes stages, dependencies, and execution parameters. The structure typically includes a stages array where each stage defines its inputs, outputs, and execution steps.

stages:
  - name: data_ingestion
    inputs:
      - source: s3://bucket/data.csv
    outputs:
      - path: /tmp/raw_data.csv
    command: |
      python -c "
        import pandas as pd
        df = pd.read_csv('s3://bucket/data.csv')
        df.to_csv('/tmp/raw_data.csv', index=False)
      "
  - name: model_training
    inputs:
      - source: /tmp/raw_data.csv
    outputs:
      - path: /models/model.pkl
    command: |
      python train.py --input /tmp/raw_data.csv --output /models/model.pkl

Pipelines can reference external resources using environment variables or configuration maps. The command field supports shell script execution with full access to system tools and installed libraries.

Managing Pipeline Dependencies

Dependencies between pipeline stages are specified explicitly in the definition file. A stage may depend on outputs from one or more previous stages. If a dependent stage fails, subsequent stages will not execute until the dependency resolves.

stages:
  - name: preprocessing
    outputs:
      - path: /tmp/processed_data.csv
    command: |
      python preprocess.py --input /data/raw.csv --output /tmp/processed_data.csv
  - name: training
    inputs:
      - source: /tmp/processed_data.csv
    command: |
      python train.py --input /tmp/processed_data.csv --output /models/model.pkl

The pipeline engine ensures that dependencies are satisfied before executing a stage. If a required input file is missing or has incorrect permissions, the stage will fail with an error like FileNotFoundError: [Errno 2] No such file or directory.

Running Pipelines

Pipelines are executed using a command-line interface. The basic invocation requires specifying the pipeline definition file and optionally a runtime environment.

pipeline run --definition /path/to/pipeline.yaml --environment production

The --definition flag points to a YAML or JSON file containing the pipeline structure. If the definition file is malformed, the system will return an error like YAML parsing error: mapping values are not allowed here.

Pipeline execution can be monitored using a status command that shows current progress and any errors encountered.

pipeline status --run-id abc123

Handling Pipeline Failures

Pipelines have built-in retry mechanisms. When a stage fails, it can be retried up to a specified number of times before failing the entire pipeline.

stages:
  - name: data_processing
    retries: 3
    command: |
      python process.py --input /data/input.csv

If a stage reaches its retry limit, the pipeline will halt and mark itself as failed. The system logs all retry attempts with timestamps and exit codes.

A stage can define an error handler that runs when the stage fails. This handler might send alerts or clean up resources.

stages:
  - name: data_upload
    on_failure:
      command: |
        echo "Upload failed, sending alert" | mail -s "Pipeline failure" [email protected]

Versioning and Configuration

Pipelines support versioning through configuration management. Each pipeline definition is stored in a version-controlled system like Git. The system tracks changes to the pipeline structure and allows rolling back to previous versions.

pipeline version list --pipeline-name data_pipeline

This command returns a list of all versions with their creation timestamps and commit hashes.

Pipeline configurations can be parameterized using environment variables or external configuration files. Parameters allow reusing the same pipeline definition across different environments.

stages:
  - name: model_evaluation
    inputs:
      - source: /models/model.pkl
    parameters:
      threshold: 0.8
    command: |
      python evaluate.py --model /models/model.pkl --threshold {{ threshold }}

Scheduling and Automation

Pipelines can be scheduled to run at regular intervals using cron-style expressions. The scheduler maintains a queue of pipeline runs and executes them based on the specified schedule.

pipeline schedule create --pipeline-name data_pipeline --cron "0 2 * * *" --environment production

The system validates cron expressions and returns an error if they are malformed. A typical error message is Invalid cron expression: Expected 5 fields, found 6.

Scheduled pipelines can be paused or resumed using specific commands.

pipeline schedule pause --schedule-id abc123

Monitoring Pipeline Execution

Pipeline execution logs are stored in a structured format and can be accessed through the command line or API. Each stage logs its start time, end time, exit code, and any standard output or error messages.

pipeline logs --run-id abc123

The system provides real-time updates during pipeline execution. Logs are streamed to stdout or saved to persistent storage for later analysis.

Resource Management

Pipelines can specify resource requirements for each stage. These include CPU, memory, and storage allocation.

stages:
  - name: model_training
    resources:
      cpu: "2"
      memory: "4Gi"
    command: |
      python train.py --input /data/input.csv

If a stage's resource request cannot be satisfied, the pipeline will fail with an error indicating resource constraints. The system checks available resources before scheduling and reports any mismatches.

Pipeline Artifacts

Pipeline stages can produce artifacts that are stored for later use. Artifacts are defined by output paths and can be accessed by subsequent stages in the same pipeline or by other pipelines.

stages:
  - name: data_processing
    outputs:
      - path: /tmp/processed_data.csv
        artifact: processed_data

Artifacts are managed through a storage system that ensures their availability. If an artifact is not found, the pipeline will fail with an error like Artifact 'processed_data' not found.

Pipeline Validation

Before execution, pipelines can be validated to check for structural issues or missing dependencies.

pipeline validate --definition /path/to/pipeline.yaml

The validation process checks that all required fields are present, dependencies are correctly specified, and commands are syntactically valid. It returns a list of errors if any are found.

Pipeline Deployment

Deployed pipelines can be managed through the command-line interface. The system supports deploying pipeline definitions to different environments like staging or production.

pipeline deploy --definition /path/to/pipeline.yaml --environment production

Deployment creates a new version of the pipeline that can be scheduled or run immediately. The system ensures that all dependencies are resolved before deployment completes.

Rollback and Recovery

When a deployed pipeline fails, it can be rolled back to a previous version using a rollback command.

pipeline rollback --pipeline-name data_pipeline --version 2

The rollback process restores the pipeline definition to its previous state. If a rollback is not possible, the system will return an error indicating that the target version does not exist.

Pipeline Cleanup

Pipelines can define cleanup procedures for stages that should be run after execution completes. This includes removing temporary files or releasing resources.

stages:
  - name: data_processing
    cleanup:
      command: |
        rm -rf /tmp/processed_data.csv

Cleanup commands are executed even if the main stage fails, ensuring that temporary resources are properly managed.

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.