Parameterizing Data Pipelines: Building Reusable, Config-Driven Workflows for Multiple Environments

Introduction

You've built a data pipeline that ingests raw sales transactions from an S3 bucket, transforms them, and loads the results into a Snowflake data warehouse. It works beautifully in development. Then your team asks you to deploy it to staging, and suddenly you're hunting through the codebase swapping out bucket names, connection strings, schema names, and batch size limits. You get staging working, and then production needs a slightly different configuration — different credentials, a larger batch size, an additional alerting webhook. Before long, you have three copies of the same pipeline, diverging slowly and silently, each one a maintenance liability.

This is the parameterization problem, and it's one of the most underestimated challenges in production data engineering. The solution isn't exotic — it's disciplined design. A well-parameterized pipeline separates what the pipeline does from where and how it runs. You define behavior once, then drive it with configuration at runtime. The result is a codebase that's easier to test, safer to deploy, and dramatically easier to maintain as your infrastructure scales.

By the end of this lesson, you'll be able to design and implement config-driven pipelines that work cleanly across multiple environments without code duplication or fragile search-and-replace deployments.

What you'll learn:

• How to identify everything in a pipeline that should be a parameter versus hardcoded logic
• Structuring YAML/JSON configuration files for multi-environment deployments
• Injecting configuration at runtime using environment variables, config files, and CLI arguments — and when to use each approach
• Validating and documenting your configuration schema so your pipelines fail loudly when misconfigured
• Building a real-world parameterized ETL pipeline that runs against dev, staging, and production environments using the same codebase

---

Prerequisites

You should be comfortable writing Python ETL scripts, familiar with the concept of environment variables, and have at least some experience with a pipeline orchestration tool (Airflow, Prefect, or similar). You don't need to be an expert in any of these — but this lesson won't explain what a DAG is or why you'd use Snowflake.

---

What Actually Belongs in a Parameter

Before writing a single config file, you need to develop a sharp instinct for what should be a parameter. The rule of thumb is this: if it changes between environments, or if changing it changes where the pipeline runs rather than what the pipeline does, it's a parameter.

Let's get concrete. Imagine a pipeline that pulls customer orders from a Postgres database, enriches them with product metadata from an API, and writes the results to a data warehouse. Here's a first pass at what's environment-specific:

Infrastructure identifiers:

• Database host, port, database name
• Schema and table names (your dev schema is probably orders_dev, not orders)
• S3 bucket names and key prefixes
• Data warehouse connection strings and warehouse sizes

Operational settings:

• Batch sizes (you might run 1,000 rows in dev, 500,000 in production)
• Retry limits and retry delays
• Timeout values
• Log levels (DEBUG in dev, WARNING in production)

Business logic knobs (the trickier category):

• Lookback windows ("process the last N days of orders")
• Feature flags ("include returns in revenue calculation: true/false")
• Rate limits for external API calls

Secrets (a special subcategory):

• Passwords, API keys, tokens — these are parameters, but they should never live in a config file that gets committed to source control. We'll handle them separately.

What shouldn't be a parameter? The core transformation logic. If you're computing gross_revenue = quantity * unit_price - discount, that formula belongs in code. If you're filtering out test accounts by checking a boolean flag in the source data, that's logic. Parameters control where and at what scale your pipeline runs; they don't replace business rules.

A mistake practitioners often make is over-parameterizing. If you find yourself with a config key called filter_test_accounts_column_name, you've probably gone too far — that's embedding business logic in config instead of code, which makes the system harder to reason about, not easier.

---

Structuring Your Configuration Files

Once you've identified your parameters, you need a home for them. The most practical structure for multi-environment pipelines is a layered config system: a base configuration that defines defaults, with environment-specific files that override only what's different.

Here's a realistic directory layout for a pipeline project:

pipeline_project/
├── config/
│   ├── base.yaml
│   ├── dev.yaml
│   ├── staging.yaml
│   └── production.yaml
├── pipeline/
│   ├── __init__.py
│   ├── extract.py
│   ├── transform.py
│   ├── load.py
│   └── config_loader.py
├── tests/
│   └── test_config.py
├── .env.example
└── run_pipeline.py

The base.yaml defines the full shape of your configuration — every key your pipeline might look for, with reasonable defaults:

# config/base.yaml

pipeline:
  name: "customer_orders_etl"
  version: "2.1.0"
  batch_size: 10000
  max_retries: 3
  retry_delay_seconds: 30
  log_level: "INFO"

source:
  type: "postgres"
  host: "localhost"
  port: 5432
  database: "orders_db"
  schema: "public"
  table: "raw_orders"
  lookback_days: 7

enrichment:
  api_base_url: "https://api.products.internal"
  request_timeout_seconds: 10
  rate_limit_calls_per_minute: 60

destination:
  type: "snowflake"
  account: "my_org.us-east-1"
  database: "ANALYTICS"
  schema: "ORDERS"
  table: "CUSTOMER_ORDERS"
  warehouse: "TRANSFORM_XS"
  write_mode: "append"

notifications:
  slack_webhook_enabled: false
  alert_on_row_count_below: 100

Now, dev.yaml only contains what's different from base:

# config/dev.yaml

pipeline:
  batch_size: 1000
  log_level: "DEBUG"

source:
  host: "dev-postgres.internal"
  schema: "orders_dev"

destination:
  schema: "ORDERS_DEV"
  warehouse: "TRANSFORM_XS"
  write_mode: "overwrite"

notifications:
  slack_webhook_enabled: false

And production.yaml overrides for the live environment:

# config/production.yaml

pipeline:
  batch_size: 500000
  max_retries: 5
  retry_delay_seconds: 60

source:
  host: "prod-postgres.internal"
  schema: "orders_prod"

destination:
  schema: "ORDERS_PROD"
  warehouse: "TRANSFORM_L"
  write_mode: "append"

notifications:
  slack_webhook_enabled: true
  alert_on_row_count_below: 10000

Notice that production.yaml doesn't touch api_base_url — the enrichment API is the same across all environments in this case. Only override what's actually different. This is important: if you copy the entire base config into each environment file, you lose the benefit of layering. Now when you need to change a shared default, you have to find and update it in multiple places.

Writing the Config Loader

The loader is the piece that reads your files and merges them. Python's PyYAML and a recursive dictionary merge function are all you need:

# pipeline/config_loader.py

import os
import yaml
from pathlib import Path
from typing import Any


def deep_merge(base: dict, override: dict) -> dict:
    """
    Recursively merge override into base.
    Keys in override take precedence; nested dicts are merged, not replaced.
    """
    result = base.copy()
    for key, value in override.items():
        if key in result and isinstance(result[key], dict) and isinstance(value, dict):
            result[key] = deep_merge(result[key], value)
        else:
            result[key] = value
    return result


def load_config(environment: str, config_dir: str = None) -> dict:
    """
    Load and merge base config with environment-specific overrides.
    
    Args:
        environment: One of 'dev', 'staging', 'production'
        config_dir: Path to config directory. Defaults to project root /config.
    
    Returns:
        Merged configuration dictionary.
    
    Raises:
        FileNotFoundError: If base.yaml or the environment config file is missing.
        ValueError: If environment is not a recognized value.
    """
    valid_environments = {"dev", "staging", "production"}
    if environment not in valid_environments:
        raise ValueError(
            f"Unknown environment '{environment}'. "
            f"Must be one of: {', '.join(sorted(valid_environments))}"
        )

    if config_dir is None:
        # Default: two levels up from this file, then into /config
        config_dir = Path(__file__).parent.parent / "config"
    else:
        config_dir = Path(config_dir)

    base_path = config_dir / "base.yaml"
    env_path = config_dir / f"{environment}.yaml"

    if not base_path.exists():
        raise FileNotFoundError(f"Base config not found at: {base_path}")
    if not env_path.exists():
        raise FileNotFoundError(
            f"Environment config for '{environment}' not found at: {env_path}"
        )

    with open(base_path) as f:
        base_config = yaml.safe_load(f)

    with open(env_path) as f:
        env_config = yaml.safe_load(f) or {}

    return deep_merge(base_config, env_config)

The deep_merge function is worth understanding carefully. A naive {**base, **override} merge would replace entire nested dictionaries. If base has 10 keys under destination and your dev.yaml overrides just schema, a shallow merge would wipe out the other 9 keys. Deep merge handles this correctly — it descends into nested dicts and only replaces individual leaf values.

---

Injecting Configuration at Runtime

Config files handle structural settings, but you have three additional mechanisms for getting values into a pipeline at runtime: environment variables, CLI arguments, and secrets managers. Each has a specific role.

Environment Variables: The Right Tool for Secrets

Passwords, API keys, and connection tokens should never live in YAML files, especially not ones that get committed to a repository. The standard approach is to read sensitive values from environment variables at runtime, and store them in a secrets manager or CI/CD secret store.

# pipeline/config_loader.py (continued)

import os


def inject_secrets(config: dict) -> dict:
    """
    Overlay secret values from environment variables into the config.
    Environment variables take precedence over any config file values.
    
    Expected env vars:
        ORDERS_DB_PASSWORD      - Postgres connection password
        SNOWFLAKE_PASSWORD      - Snowflake user password
        PRODUCTS_API_KEY        - Bearer token for enrichment API
        SLACK_WEBHOOK_URL       - Slack notification URL (optional)
    """
    secrets_map = {
        ("source", "password"): "ORDERS_DB_PASSWORD",
        ("destination", "password"): "SNOWFLAKE_PASSWORD",
        ("enrichment", "api_key"): "PRODUCTS_API_KEY",
        ("notifications", "slack_webhook_url"): "SLACK_WEBHOOK_URL",
    }

    result = config.copy()

    for key_path, env_var in secrets_map.items():
        value = os.environ.get(env_var)
        if value is not None:
            # Navigate to the correct nested key and set it
            d = result
            for part in key_path[:-1]:
                d = d.setdefault(part, {})
            d[key_path[-1]] = value

    return result

Security tip: Add a startup check that verifies required secrets are present before the pipeline does any real work. A missing SNOWFLAKE_PASSWORD that surfaces 45 minutes into a pipeline run is much more painful than one that fails immediately at launch.

def validate_required_secrets(config: dict) -> None:
    """Fail fast if required secrets are not present."""
    required = [
        (("source", "password"), "ORDERS_DB_PASSWORD"),
        (("destination", "password"), "SNOWFLAKE_PASSWORD"),
        (("enrichment", "api_key"), "PRODUCTS_API_KEY"),
    ]
    missing = []
    for key_path, env_var in required:
        d = config
        try:
            for part in key_path:
                d = d[part]
        except KeyError:
            missing.append(env_var)
    
    if missing:
        raise EnvironmentError(
            f"Required environment variables not set: {', '.join(missing)}"
        )

CLI Arguments: Environment Selection and Run-Time Overrides

The pipeline entry point should accept the environment as a command-line argument, not read it from another environment variable (that gets confusing fast). Use Python's argparse for this:

# run_pipeline.py

import argparse
import logging
from pipeline.config_loader import load_config, inject_secrets, validate_required_secrets
from pipeline.extract import extract_orders
from pipeline.transform import transform_orders
from pipeline.load import load_to_warehouse


def parse_args():
    parser = argparse.ArgumentParser(
        description="Customer Orders ETL Pipeline",
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
    )
    parser.add_argument(
        "--env",
        required=True,
        choices=["dev", "staging", "production"],
        help="Target environment for this pipeline run",
    )
    parser.add_argument(
        "--lookback-days",
        type=int,
        default=None,
        help="Override the number of lookback days (useful for backfills)",
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Run extraction and transformation but skip the load step",
    )
    return parser.parse_args()


def main():
    args = parse_args()

    # Load and merge config
    config = load_config(args.env)
    config = inject_secrets(config)
    validate_required_secrets(config)

    # Apply CLI overrides on top of file-based config
    if args.lookback_days is not None:
        config["source"]["lookback_days"] = args.lookback_days

    # Configure logging from config
    logging.basicConfig(
        level=getattr(logging, config["pipeline"]["log_level"]),
        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    )
    logger = logging.getLogger("pipeline")
    logger.info(
        f"Starting {config['pipeline']['name']} v{config['pipeline']['version']} "
        f"in {args.env} environment"
    )

    # Run the pipeline
    raw_data = extract_orders(config)
    transformed_data = transform_orders(raw_data, config)

    if args.dry_run:
        logger.info(f"Dry run complete. {len(transformed_data)} rows would be loaded.")
    else:
        load_to_warehouse(transformed_data, config)


if __name__ == "__main__":
    main()

This layering — base config, environment overrides, secrets injection, CLI overrides — gives you a predictable precedence chain. The most specific source wins. That's a mental model your whole team can learn quickly.

---

Validating Your Configuration Schema

Silently ignoring a missing or mistyped config key is one of the most frustrating bugs to track down in production. Your pipeline runs fine until it hits the code path that reads the bad key, and by then you may have already spent 20 minutes on partially processed data.

The cleanest way to enforce your config schema in Python is with pydantic. Define your configuration as typed models and validate on load:

# pipeline/config_schema.py

from pydantic import BaseModel, Field, field_validator
from typing import Literal
from enum import Enum


class WriteMode(str, Enum):
    append = "append"
    overwrite = "overwrite"
    merge = "merge"


class PipelineConfig(BaseModel):
    name: str
    version: str
    batch_size: int = Field(gt=0, le=1_000_000)
    max_retries: int = Field(ge=0, le=10)
    retry_delay_seconds: int = Field(ge=0)
    log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = "INFO"


class SourceConfig(BaseModel):
    type: Literal["postgres", "mysql"]
    host: str
    port: int = Field(gt=0, lt=65536)
    database: str
    schema: str
    table: str
    lookback_days: int = Field(gt=0, le=365)
    password: str = ""  # Populated at runtime from env vars


class EnrichmentConfig(BaseModel):
    api_base_url: str
    request_timeout_seconds: int = Field(gt=0, le=300)
    rate_limit_calls_per_minute: int = Field(gt=0, le=6000)
    api_key: str = ""  # Populated at runtime from env vars

    @field_validator("api_base_url")
    @classmethod
    def must_be_https(cls, v):
        if not v.startswith("https://"):
            raise ValueError(
                f"api_base_url must use HTTPS. Got: {v}"
            )
        return v


class DestinationConfig(BaseModel):
    type: Literal["snowflake", "bigquery", "redshift"]
    account: str
    database: str
    schema: str
    table: str
    warehouse: str
    write_mode: WriteMode = WriteMode.append
    password: str = ""  # Populated at runtime from env vars


class NotificationConfig(BaseModel):
    slack_webhook_enabled: bool = False
    slack_webhook_url: str = ""
    alert_on_row_count_below: int = Field(ge=0)


class AppConfig(BaseModel):
    pipeline: PipelineConfig
    source: SourceConfig
    enrichment: EnrichmentConfig
    destination: DestinationConfig
    notifications: NotificationConfig


def validate_config(raw_config: dict) -> AppConfig:
    """
    Parse and validate raw config dict against the schema.
    Raises pydantic.ValidationError with detailed messages on failure.
    """
    return AppConfig(**raw_config)

Now update run_pipeline.py to validate before running:

from pipeline.config_schema import validate_config, AppConfig

# After loading and injecting secrets:
config_dict = inject_secrets(config_dict)
config: AppConfig = validate_config(config_dict)  # Raises on bad config

# Access with type safety
batch_size = config.pipeline.batch_size  # int, guaranteed

The payoff here is significant. When someone accidentally sets batch_size: "ten thousand" in a YAML file, Pydantic catches it immediately with a message like batch_size: value is not a valid integer. When a new team member adds a staging config that points to an HTTP (not HTTPS) enrichment endpoint, the must_be_https validator fires before a single byte of data is processed.

Practical note: If you're using Python 3.9 or earlier, use pydantic v1 syntax (@validator instead of @field_validator). The code above targets pydantic v2.

---

Using Configuration Inside Pipeline Components

The config object needs to flow through your pipeline cleanly. The most practical pattern is to pass the config (or the relevant sub-config) explicitly to each component function. Avoid module-level globals or singletons — they make testing painful.

# pipeline/extract.py

import logging
import psycopg2
import pandas as pd
from datetime import datetime, timedelta
from pipeline.config_schema import AppConfig

logger = logging.getLogger(__name__)


def extract_orders(config: AppConfig) -> pd.DataFrame:
    """
    Extract recent orders from the source Postgres database.
    Batch size and lookback window are driven entirely by config.
    """
    src = config.source
    pipeline = config.pipeline
    
    cutoff_date = datetime.utcnow() - timedelta(days=src.lookback_days)
    
    logger.info(
        f"Extracting orders from {src.host}/{src.database}.{src.schema}.{src.table} "
        f"since {cutoff_date.strftime('%Y-%m-%d')}, "
        f"batch size: {pipeline.batch_size:,}"
    )

    conn = psycopg2.connect(
        host=src.host,
        port=src.port,
        dbname=src.database,
        user="pipeline_user",  # username is not secret; could also be in config
        password=src.password,
    )

    query = """
        SELECT
            order_id,
            customer_id,
            product_id,
            quantity,
            unit_price,
            discount_amount,
            order_timestamp,
            status
        FROM {schema}.{table}
        WHERE order_timestamp >= %(cutoff)s
        AND status != 'TEST'
        ORDER BY order_timestamp
        LIMIT %(batch_size)s
    """.format(schema=src.schema, table=src.table)

    df = pd.read_sql(
        query,
        conn,
        params={"cutoff": cutoff_date, "batch_size": pipeline.batch_size},
    )
    
    logger.info(f"Extracted {len(df):,} rows")
    conn.close()
    return df

Notice that status != 'TEST' is hardcoded in the SQL — that's intentional. Filtering test orders is business logic, not a configurable behavior. The schema name, table name, lookback window, and batch size are all drawn from config. This is the boundary we talked about earlier.

---

Integrating with Orchestrators: Airflow as a Case Study

If you're running pipelines in Airflow, you have an additional configuration layer to think about: Airflow Variables and Connections. The pattern that works best in practice is to use Airflow's environment-specific deployment (separate Airflow instances for dev/staging/prod, or namespaced Variables), and pass the environment name into your DAG at runtime.

# dags/customer_orders_dag.py

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.models import Variable
from datetime import datetime, timedelta

# Pull environment from Airflow Variables, not hardcoded
ENVIRONMENT = Variable.get("pipeline_environment", default_var="dev")


def run_extraction(**context):
    from pipeline.config_loader import load_config, inject_secrets
    from pipeline.config_schema import validate_config
    from pipeline.extract import extract_orders

    config_dict = load_config(ENVIRONMENT)
    config_dict = inject_secrets(config_dict)
    config = validate_config(config_dict)

    # Handle backfill: use execution_date as the reference point
    execution_date = context["execution_date"]
    days_since_execution = (datetime.utcnow() - execution_date).days
    if days_since_execution > config.source.lookback_days:
        config.source.lookback_days = days_since_execution + 1

    df = extract_orders(config)
    # Store row count for downstream tasks to validate
    context["ti"].xcom_push(key="extracted_row_count", value=len(df))
    # Serialize to parquet and store path in XCom for the next task
    temp_path = f"/tmp/orders_{execution_date.strftime('%Y%m%d')}.parquet"
    df.to_parquet(temp_path, index=False)
    context["ti"].xcom_push(key="staging_file_path", value=temp_path)


default_args = {
    "owner": "data-engineering",
    "retries": 2,
    "retry_delay": timedelta(minutes=5),
    "email_on_failure": True,
    "email": ["data-alerts@yourcompany.com"],
}

with DAG(
    dag_id="customer_orders_etl",
    default_args=default_args,
    schedule_interval="0 6 * * *",  # 6 AM UTC daily
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=["orders", "etl", ENVIRONMENT],
) as dag:

    extract_task = PythonOperator(
        task_id="extract_orders",
        python_callable=run_extraction,
        provide_context=True,
    )
    # ... additional tasks for transform and load

Airflow tip: Don't use Variable.get() at module level in production DAGs on large Airflow deployments — it queries the metadata database every time the scheduler parses the DAG file. Instead, use it inside the callable function, or set the variable in the DAG's params and access via the context. The example above is simplified for readability.

---

Hands-On Exercise: Build a Multi-Environment Pipeline

Let's put this together. You're going to build a parameterized pipeline that processes daily website clickstream data. It reads from a CSV file (simulating an S3 download), filters and aggregates the data, and writes a summary to a SQLite database (simulating a data warehouse).

Setup: Create the project structure shown above. Install dependencies:

pip install pyyaml pydantic pandas

Step 1: Create your base config.

# config/base.yaml
pipeline:
  name: "clickstream_aggregator"
  version: "1.0.0"
  batch_size: 50000
  log_level: "INFO"

source:
  type: "csv"
  input_directory: "./data/raw"
  file_pattern: "clicks_*.csv"
  lookback_days: 1

destination:
  type: "sqlite"
  database_path: "./data/warehouse.db"
  table: "daily_page_views"
  write_mode: "append"

filtering:
  exclude_bot_traffic: true
  minimum_session_duration_seconds: 5

Step 2: Create dev and production overrides.

# config/dev.yaml
pipeline:
  log_level: "DEBUG"
  batch_size: 1000

source:
  input_directory: "./data/dev/raw"
  lookback_days: 3

destination:
  database_path: "./data/dev/warehouse.db"
  table: "daily_page_views_dev"
  write_mode: "overwrite"

# config/production.yaml
source:
  input_directory: "/mnt/data/clickstream/raw"
  lookback_days: 1

destination:
  database_path: "/mnt/data/warehouse/analytics.db"
  write_mode: "append"

Step 3: Implement the config loader using the load_config and deep_merge functions from earlier in this lesson.

Step 4: Write the pipeline components. Your extract.py should read CSV files from config.source.input_directory matching config.source.file_pattern. Your transform.py should apply the filtering settings from config.filtering. Your load.py should write to the SQLite path in config.destination.database_path.

Step 5: Test your config isolation. Run:

python run_pipeline.py --env dev
python run_pipeline.py --env production --dry-run
python run_pipeline.py --env dev --lookback-days 7

Verify that dev runs write to the dev database, that the dry run skips the load step, and that the lookback override changes the date range of your query.

Challenge extension: Add a --config-override CLI argument that accepts key=value pairs in dot notation (e.g., --config-override pipeline.batch_size=2000) and applies them on top of everything else. This is genuinely useful for one-off backfills.

---

Common Mistakes & Troubleshooting

Shallow Merging Config Dictionaries

The mistake: Using {**base_config, **env_config} to merge configs. If your environment YAML overrides a single nested key, the entire parent dictionary gets replaced.

The symptom: You add write_mode: overwrite to dev.yaml under destination, and suddenly your pipeline can't find destination.schema anymore.

The fix: Always use a recursive deep merge, as shown in the deep_merge function above. Write a unit test that confirms a single nested key override doesn't destroy sibling keys.

Committing Secrets to Config Files

The mistake: Putting a real password in dev.yaml because "it's just dev." Dev configs get committed, repos get cloned to new machines, and secrets proliferate.

The fix: Use environment variables for all secrets without exception. Check your .gitignore to make sure .env files are excluded. Scan your repo with a tool like git-secrets or truffleHog as part of your CI pipeline.

Silent Fallback to Default Values

The mistake: Using config.get("batch_size", 10000) everywhere instead of a schema validator. When batch_size is misspelled as batch-size in a YAML file, you silently use the default and your pipeline runs at a tenth of the expected speed. No errors, just slow and inexplicable behavior.

The fix: Validate your config against a schema immediately after loading. Pydantic's strict mode will catch key mismatches. If a key is optional with a default, make that explicit in the schema rather than scattered across dict.get() calls throughout the codebase.

Environment Bleed in Test Suites

The mistake: Your test suite reads from environment variables or the current ENV setting, which means tests behave differently on a developer's laptop versus CI.

The fix: In your tests, always construct config explicitly:

# tests/test_transform.py

def get_test_config():
    """Build a minimal config for testing — never reads from disk or env vars."""
    return {
        "pipeline": {
            "name": "test_pipeline",
            "batch_size": 100,
            "log_level": "WARNING",
            # ... other required fields
        },
        # ...
    }

def test_transform_excludes_bot_traffic():
    config = validate_config(get_test_config())
    input_df = pd.DataFrame({...})  # synthetic test data
    result = transform_orders(input_df, config)
    assert not result["is_bot"].any()

YAML Type Coercion Surprises

The mistake: Writing log_level: DEBUG (no quotes) in a YAML file. YAML will parse this as the string "DEBUG", which is fine — until someone writes port: 05432, which YAML parses as the octal number 2842 in some parsers.

The fix: Quote strings that could be ambiguous. Port numbers, version strings, and anything that looks like it could be a number should be unambiguous. Pydantic validation will catch type errors downstream, but explicit YAML is easier to read.

Config Drift Across Environments

The mistake: Over time, your environment configs diverge. Staging has settings that production doesn't, production has tuning parameters that nobody documented. Eventually, "works in staging" stops meaning anything.

The fix: Keep environment config files minimal — only genuine overrides from base. Review them in code review the same way you'd review application code. Add a CI check that validates all environment configs can be successfully loaded and pass schema validation.

---

Summary & Next Steps

You now have a complete mental model and implementation pattern for config-driven pipelines. Let's recap the key principles:

Separate concerns cleanly. Pipeline logic lives in code. Environment-specific behavior lives in config. Secrets live in environment variables and secrets managers, never in files.

Layer your configuration. Base defaults, environment overrides, runtime injection. Each layer only specifies what it owns. The most specific source wins.

Validate early and loudly. A schema validator like Pydantic transforms mysterious runtime failures into clear, immediate error messages at startup. This pays dividends every single time a config file is edited.

Pass config explicitly. Thread your config object through your pipeline components as a function argument. It makes testing trivial and makes the data flow obvious.

Keep environment files minimal. The more you override in an environment file, the harder it is to understand how that environment actually differs from base. Override only what genuinely changes.

Where to Go From Here

With parameterized pipelines under your belt, the natural next topics to explore are:

• Secrets management at scale: Integrating with HashiCorp Vault, AWS Secrets Manager, or GCP Secret Manager instead of relying on environment variables set manually. The config injection pattern you learned here makes this a straightforward swap.

• Dynamic configuration in Airflow: Using Airflow's Params API to pass runtime configuration through to task operators without hardcoding it in the DAG file.

• Configuration versioning and auditing: Storing a hash or snapshot of the config used for each pipeline run alongside the run's output, so you can always answer "what config produced this data?"

• Feature flags for data pipelines: Using a lightweight feature flag system (LaunchDarkly, Unleash, or even a simple database table) to control pipeline behavior without deployments — particularly useful for gradual rollouts of new transformation logic.

The ability to confidently promote a pipeline from dev through staging to production — knowing that the only things changing are the config, not the code — is one of the most tangible signs of a mature data engineering practice. You've now got the foundation to build exactly that.