Data Pipeline Design Pattern

A battle-tested architecture for building maintainable, configuration-driven data processing pipelines in Python with pandas. The pattern has been validated in production systems across ETL workflows, rule engines, report generators, and multi-source data fusion projects.

How to Use This Skill (怎么用)

This Skill is a design guide, not a code library. Just describe your data processing needs in natural language — no special commands or parameters required. For example:

| You say | Skill does | |---------|------------| | "帮我做一个销售提成计算系统" | Guides you through 5-stage pipeline design | | "我有 3 个 Excel 要合并计算" | Designs a configuration-driven ETL project | | "设计一个规则引擎" | Applies strategy pattern + config-driven architecture | | "数据处理流水线怎么做" | Walks through all 14 design principles |

What you need to provide: A brief description of your data, business rules, and output format. The Skill will ask clarifying questions before writing any code.

When to Use vs Not Use

Use this Skill when:

| Scenario | Example | |----------|---------| | Rule-based calculation | Commission, pricing, scoring systems where rules change frequently | | Multi-source ETL | Merging data from 5+ Excel/CSV sources with complex join logic | | Report generation | Periodic reports needing aggregation, KPI calculation | | Data enrichment | Adding business labels, classifications, external data to raw records | | Quality validation | Automated checks on output data: completeness, consistency, domain |

Do NOT use when:

• Single-script transformations (< 100 lines)
• Real-time streaming data
• Business rules are static and simple
• Data volume exceeds millions of rows (this pattern uses pandas, not Spark/Dask; for big data, consider distributed frameworks instead)

Before You Start: Requirement Clarification (需求澄清)

When a user triggers this Skill, do NOT jump straight to coding. Ask these questions first to confirm the requirement. Adapt the questions to the user's context — skip those already answered:

A. Project Type

• What kind of project is this? (ETL / rule calculation / report / multi-source fusion)
• One-time script or recurring pipeline?

B. Data Sources

• How many input data sources? File formats? (Excel / CSV / database / API)
• Sample data available? (if yes, read it to understand the schema)
• Which columns are the join keys between sources?

C. Business Logic

• What are the calculation rules? Formulas, thresholds, conditions?
• How often do these rules change? (frequently → configuration-driven is critical)
• Any special cases? (violations, exceptions, different paths for different data types)

D. Output

• What should the final output look like? (Excel / HTML report / database write)
• Need data quality validation? Which checks?
• Need visualization? (charts, KPI cards, summary tables) — this is optional and can be handled by other Skills (HTML design, xlsx formatting, BI tools)

E. Scale & Environment

• Rough data volume? (hundreds / tens of thousands / millions of rows)
• Single environment or dev/staging/prod?

After confirming answers, propose a pipeline structure matching the user's scenario, then start implementation.

Core Design Principles

1. Configuration-Driven Architecture (配置驱动)

The most important principle. All business rules, field names, file paths, thresholds, and matching logic live in config/ as pure Python dictionaries and constants — never hardcoded in business logic.

config/
├── paths.py            # File paths, sheet names, output filenames
├── matching_config.py  # Join/match rules, lookup configs
├── calc_config.py      # Calculation formulas, thresholds, field lists
└── output_config.py    # Report settings, column mappings

Why this matters: When business rules change (e.g., threshold moves from 7 days to 14 days, or a field gets renamed), only the config file changes — zero logic code modifications.

Config format choice:

| Format | Best for | |--------|----------| | Python dict (recommended) | Pure Python team, complex config with lots of fields, IDE autocomplete | | YAML file | Non-technical editors, cross-language sharing, environment layering (dev/prod) | | JSON file | Strict schema validation, sharing with frontend/API |

Both Python dict and YAML are valid choices — pick based on your team. The Skill shows both patterns.

Config structure rule: Every join/match operation uses a unified config dict:

MATCH_RULE = {
    "left_keys": ["column1", "column2"],   # Left table join keys
    "right_keys": ["column1", "column2"],  # Right table join keys
    "select_cols": ["target_col"],         # Columns to extract from right table
    "rename_cols": {"old": "new"},         # Optional column rename after match
    "default_value": "default",            # Optional fallback for unmatched rows
}

2. Pipeline Stage Pattern (流水线阶段)

Organize processing as discrete stages controlled by a simple list. Each stage produces output consumed by the next:

STAGES = ["enrich", "calc"]

if "enrich" in STAGES:
    df_enriched = run_enrich()
    save_output(df_enriched, "enriched")

if "calc" in STAGES:
    df = run_calculation()
    # → quality checks → extract → report

This allows selective execution during development or debugging — run only the stages needed.

3. Incremental Data Enrichment (增量数据富化)

Each transformation adds new columns to the DataFrame — never removes rows. Enrichments compose sequentially:

class DataEnricher:
    def __init__(self, df: pd.DataFrame):
        self.df = df

    def add_row_index(self) -> "DataEnricher":    # return self for chaining
        self.df.insert(0, "row_id", range(1, len(self.df) + 1))
        return self

    def add_time_period(self) -> "DataEnricher":
        self.df["year"] = self.df["timestamp"].dt.year
        return self

    def enrich_from_lookup(self) -> "DataEnricher":
        cfg = LOOKUP_MATCH_CONFIG
        df_lookup = load_lookup_sheet("category_lookup")
        self.df = left_join_tag(self.df, df_lookup, cfg, "category_label")
        return self

Key characteristics:

• Methods return self for fluent chaining
• Each method adds columns, never removes data rows
• Order matters: later enrichments can reference columns added by earlier ones
• Common enrichments apply to all data; specialized enrichments inherit and extend

4. Strategy Pattern for Specialized Processing

Common interface with specialized implementations per dimension:

src/steps/enrich/
├── common.py          # CommonEnricher — shared logic for all data
├── source_a.py        # SourceAEnricher — source-specific logic
├── source_b.py        # SourceBEnricher — source-specific logic
└── source_c.py        # SourceCEnricher — source-specific logic

Each specialized class follows the same pattern:

• Accepts df in __init__
• Implements process() that chains enrichment methods
• All config imported from config/

5. Universal Join/Match Utilities (通用匹配工具)

A small set of well-tested utilities handle all data joining patterns:

| Function | Purpose | |----------|---------| | merge_lookup() | Generic left join with single-key map optimization | | left_join_tag() | Left join with result statistics and column rename | | merge_asof_tag() | Time-based fuzzy merge (monthly granularity) | | load_lookup_sheet() | Unified loader for xlsx/csv with path config | | merge_files() | Batch merge multiple files from a folder |

Optimization detail: merge_lookup() automatically chooses between pd.Series.map() (10x+ faster for single-key joins) and pd.merge() (for multi-key joins).

6. Data Quality Layer (数据质量)

Separate validation that runs after the main pipeline, checking:

• Column completeness (required fields exist)
• Null rates on key fields
• Value domain validation (only expected values appear)
• Statistical anomalies (extreme values, inconsistent calculations)

Each check returns a {"status": "OK/WARN/ERROR", "message": "...", "details": ...} dict, enabling structured report generation.

7. Separate Calculation from Presentation

src/
├── steps/                 # Pure data computation — no UI/formatting
│   ├── enrich/
│   ├── calculate/
│   └── extract/
├── quality/               # Validation layer
│   └── quality.py
└── output/                # Report generation — HTML, Excel output
    └── report.py

Reports are generated as self-contained HTML files (no network required) with Chart.js for visualization.

8. Non-Intrusive Logging System

A logging system integrated into match utilities that:

• Defaults to OFF — zero performance impact in normal execution
• Activated via enable_match_log(True) only during debugging
• Records first 10 rows of each match and intermediate table
• Exports to a structured Excel workbook with per-step sheets

# In left_join_tag / merge_asof_tag:
_log_match_sample(df, match_cfg, label_name, match_type)  # auto-called after match

# In load_lookup_sheet / merge_files:
_log_intermediate_table(df, table_name)  # auto-called after load

9. Symbolic Field References (符号化字段引用)

Define all column names as module-level constants — never use string literals in logic:

# calc_config.py
FIELD_NAMES = {
    "concat_col": "field_name",
    "rate_col": "rate",
    "result_col": "result",
}

This makes field name changes trivial and eliminates typos.

10. Configuration Objects as Function Parameters

Complex extraction/matching logic receives its entire configuration as a single dict:

# extraction_config.py — everything Extractor needs in one place
EXTRACTOR_CONFIG = {
    "extract_fields": [...],
    "computed_fields": [...],
    "filter_field": "category",
    "group_rules": {
        "group_a": {"source_field": "...", "value_field": "...", "split_value": True},
        "group_b": {...},
    },
    "lookup_match": {"left_keys": [...], "right_keys": [...], "select_cols": [...]},
}

# extract.py
class DataExtractor:
    def __init__(self, config: dict):
        self.cfg = config

This pattern makes the extractor fully reusable — a different data type just needs a different config dict.

11. Registry & Auto-Discovery (注册表与自动发现)

Replace if/else chains with a registry pattern for selecting implementations:

# registry.py
_REGISTRY = {}

def register(name: str):
    """Decorator to register an enricher class."""
    def wrapper(cls):
        _REGISTRY[name] = cls
        return cls
    return wrapper

def get_enricher(name: str):
    """Get enricher class by name."""
    return _REGISTRY.get(name)

# Usage in enrich modules:
@register("source_a")
class SourceAEnricher:
    ...

# In orchestrator:
enricher_cls = get_enricher(source_name)
enricher = enricher_cls(df)

This eliminates the need to update a central if/else block every time a new data source is added — just create the new class with @register("new_source").

12. Context Object (上下文对象)

Pass structured context between stages instead of bare DataFrames:

class PipelineContext:
    """Shared state across pipeline stages."""
    def __init__(self):
        self.df: pd.DataFrame = None       # Current working DataFrame
        self.stats: dict = {}              # Stage-level statistics
        self.cache: dict = {}              # Reusable intermediate results
        self.errors: list = []             # Collected errors/warnings

    def record_stat(self, stage: str, key: str, value):
        self.stats.setdefault(stage, {})[key] = value

This provides:

• Single source of truth for pipeline state
• Easy progress tracking and audit trail
• No more scattered global variables

13. Environment-Layered Configuration (环境分层配置)

Support multiple environments with config inheritance:

# config/settings.py
ENVIRONMENT = "dev"  # dev | staging | prod

BASE_PATHS = {
    "dev":     {"data": Path("D:/data/dev"),  "output": Path("D:/output/dev")},
    "staging": {"data": Path("E:/data/stage"), "output": Path("E:/output/stage")},
    "prod":    {"data": Path("F:/data/prod"),  "output": Path("F:/output/prod")},
}

CURRENT_PATHS = BASE_PATHS[ENVIRONMENT]

Switch environments by changing one variable — all paths and settings follow.

14. Idempotency & Resume (幂等与断点续跑)

Save intermediate results so individual stages can be re-run without recomputing everything:

STAGES = ["enrich", "calc"]   # Comment out completed stages

if "enrich" in STAGES:
    df = run_enrich()
    df.to_parquet("_cache/enriched.parquet")   # Save intermediate

if "calc" in STAGES:
    df = pd.read_parquet("_cache/enriched.parquet")  # Load from cache
    df = run_calculation(df)

This is especially valuable when:

• Debugging a late stage (skip the slow early stages)
• Business rules change mid-pipeline (re-run only affected stages)
• A stage fails halfway (resume from last successful cache)

15. Defensive Error Handling (防御性异常处理)

Each stage should handle common failures gracefully — never let one bad row or one missing file crash the entire pipeline.

File & Data Loading

def load_lookup_sheet(path_key: str) -> pd.DataFrame:
    """Load with graceful failure — never crash, always report."""
    cfg = DATA_SOURCES.get(path_key)
    if cfg is None:
        raise ValueError(f"Unknown data source: '{path_key}'. Available: {list(DATA_SOURCES.keys())}")

    file_path = cfg["file"]
    if not file_path.exists():
        raise FileNotFoundError(
            f"File not found: {file_path}\n"
            f"  Source key: '{path_key}'\n"
            f"  Environment: {ENVIRONMENT}\n"
            f"  Did you put the file in the right directory?"
        )

    try:
        return pd.read_excel(file_path, sheet_name=cfg.get("sheet", 0))
    except Exception as e:
        raise RuntimeError(f"Failed to read '{file_path}': {e}") from e

Missing Column Guards

# Before any join or calculation, verify expected columns exist
def _ensure_columns(df: pd.DataFrame, required: list, stage: str):
    """Check required columns exist; give actionable error if not."""
    missing = [c for c in required if c not in df.columns]
    if missing:
        available = list(df.columns)
        raise KeyError(
            f"[{stage}] Missing columns: {missing}\n"
            f"  Available columns: {available}\n"
            f"  Tip: check if an earlier stage produced these columns, "
            f"or if the source file header changed."
        )

Safe Evaluation

# Never eval() raw user input without guards
def safe_calculate(df: pd.DataFrame, formula: str, result_col: str):
    """Calculate with formula, catching common errors."""
    try:
        df[result_col] = df.eval(formula)
    except Exception as e:
        # Show what columns are actually available to help debugging
        cols = list(df.columns)
        raise RuntimeError(
            f"Formula failed: '{formula}'\n"
            f"  Error: {e}\n"
            f"  Available columns: {cols}\n"
            f"  Tip: check column names match the formula exactly (case, spaces, special chars)"
        ) from e

Stage-Level Try/Catch in run.py

# Each stage wrapped so one failure doesn't lose all progress
STAGES = ["enrich", "calc"]

results = {}
for stage in STAGES:
    try:
        print(f"\n--- Stage: {stage} ---")
        if stage == "enrich":
            df = run_enrich()
            df.to_parquet("_cache/enriched.parquet")
            results[stage] = "OK"
        elif stage == "calc":
            df = pd.read_parquet("_cache/enriched.parquet")
            df = run_calculation(df)
            results[stage] = "OK"
    except Exception as e:
        results[stage] = f"FAILED: {e}"
        print(f"\n  [{stage}] FAILED — skipping remaining stages")
        print(f"  Error: {e}")
        print(f"  Cached data at _cache/ is still valid for debugging")
        break   # Stop pipeline, but don't lose cached results

print(f"\nPipeline results: {results}")

Summary: What to Guard Against

| Failure type | Protection | |-------------|-----------| | File not found | Clear error with file path, source key, and environment info | | Missing column | List missing vs available columns, hint at possible cause | | Formula error | Show formula + available columns, not a cryptic traceback | | Bad data row | Quality check catches it; don't let one row crash everything | | Stage failure | Try/catch per stage; cached intermediate data survives |

The principle: fail loudly with actionable messages, never silently or with a cryptic stack trace the user can't understand.

Project Directory Structure Template

When creating a new project using this pattern:

project_name/
├── config/
│   ├── settings.py          # Environment, base paths
│   ├── matching_config.py   # Join/match rules
│   ├── calc_config.py       # Formulas, thresholds, field lists
│   └── output_config.py     # Report settings, column mappings
├── src/
│   ├── utils.py             # merge_lookup, load_data, common helpers
│   ├── context.py           # PipelineContext class
│   ├── registry.py          # Component registry
│   ├── steps/               # Pipeline stages
│   │   ├── enrich/          # Data enrichment
│   │   │   ├── common.py
│   │   │   ├── source_a.py
│   │   │   └── source_b.py
│   │   ├── calculate/       # Core calculation
│   │   │   └── calc.py
│   │   └── extract/         # Result extraction
│   │       └── extract.py
│   ├── quality/             # Data validation
│   │   └── quality.py
│   └── output/              # Report generation
│       └── report.py
├── _cache/                  # Intermediate data (gitignored)
├── run.py                   # Entry point with STAGES control
├── requirements.txt
└── README.md

Pipeline Variants

Adapt the stage sequence to your use case:

ETL Pipeline

Extract (load sources) → Transform (clean + enrich) → Validate → Load (save output)

Rule Calculation Pipeline

Enrich (add labels) → Match Rules (join policy tables) → Calculate (apply formulas) → Validate

Report Generation Pipeline

Aggregate (group by dimensions) → Calculate KPIs → Validate → Render (HTML/Excel)

Implementation Checklist

When implementing a new pipeline using this pattern:

1. Define configs first — paths, field names, rules, thresholds all in config/
2. Build generic utilities — merge_lookup(), load_lookup_sheet(), left_join_tag()
3. Create Context class — shared state across stages
4. Implement pipeline stages — one module per stage, each consuming previous output
5. Use the enrichment pattern — each data enrichment as a method returning self
6. Add quality validation — structured checks producing {status, message, details} dicts
7. Generate reports last — separate from calculation logic
8. Add match logging — non-intrusive, toggleable debugging aid
9. Control with STAGES — run.py as a simple controller
10. Set up registry — if you have > 3 variants of any component

Anti-Patterns to Avoid

• Hardcoding column names in logic (use constants from config)
• Hardcoding file paths (use centralized path config)
• Mixing calculation with presentation (reports live in output/, not in steps)
• One giant function doing everything (break into enrichment methods)
• Duplicating join logic (reuse merge_lookup/left_join_tag everywhere)
• Writing validation inline (separate quality module produces structured results)
• Giant if/else chains for component selection (use registry pattern)
• Scattered globals for pipeline state (use Context object)

Common Pitfalls & FAQ (常见踩坑与排错)

Pitfall 1: Merge returns 0 matches

Symptom: left_join_tag() says "matched 0 rows" when you expect matches.

Causes and fixes:

| Cause | How to check | Fix | |-------|-------------|-----| | Key column type mismatch (int vs str) | df["key"].dtype vs lookup["key"].dtype | merge_lookup() auto-converts to str — make sure you're using it, not raw pd.merge() | | Leading/trailing spaces in keys | df["key"].str.strip() shows changed values | Add .str.strip() before merge, or fix source data | | Fullwidth/halfwidth characters (e.g., Chinese brackets) | df["name"].iloc[0] looks different from lookup | Use merge_lookup() with normalize_brackets_cols | | Lookup table has duplicates for the same key | lookup[key].value_counts() | merge_lookup() auto-deduplicates; if it picks wrong row, deduplicate manually first |

Pitfall 2: Column disappears after merge

Symptom: A column you expected is gone from the DataFrame after a merge.

Cause: Both tables have the same column name (not a join key). Pandas renames one to col_x and one to col_y.

Fix: Use cleanup_merge_suffixes() (see references/utility_functions.md) to merge _x/_y back. Or only select_cols the columns you actually need from the right table.

Pitfall 3: df.eval() throws KeyError

Symptom: df.eval("amount - cost") fails with KeyError: 'amount'.

Fix: Use safe_calculate() from Principle 15. It shows the formula AND available columns, so you can spot the typo instantly.

Pitfall 4: Everything works in dev but fails in prod

Symptom: Pipeline runs fine on your machine, crashes on the server.

Common causes:

• Hardcoded path like D:/my_data/ that doesn't exist on the server
• File encoding difference (UTF-8 vs GBK on Windows)
• Different pandas version (e.g., df.agg(named_args) only in pandas 0.25+)

Fix: Use config/settings.py with ENVIRONMENT switch (Principle 13). Test with ENVIRONMENT = "prod" locally before deploying.

Pitfall 5: Pipeline runs but output is all zeros

Symptom: All calculation results are 0 or NaN.

Checklist:

1. Did the policy/rate matching actually work? Check the match log.
2. Are rate columns filled with NaN after merge? → fillna(0) before calculation.
3. Is the formula using the right column names? Print df.columns at the calculation stage.
4. Are there filter conditions accidentally excluding all rows?

Pitfall 6: Excel file won't open / corrupted output

Symptom: The output .xlsx file throws an error when opened.

Fix: Check for:

• Special characters in column names (some Excel versions reject certain chars)
• Overly long sheet names (> 31 chars)
• Column with mixed types (int and str) causing pandas to write corrupted data
• Solution: cast all output columns to str before writing

FAQ

Q: Can I use this pattern with databases instead of Excel? A: Yes. Replace load_lookup_sheet() with pd.read_sql(). All downstream code works unchanged — they only care about getting a DataFrame.

Q: What if I have 100+ data sources? A: Don't create 100 enricher classes. Group them by processing logic — if 50 sources need the same 3 join steps, they share one enricher. Use the registry pattern (Principle 11) to map source names to enrichers.

Q: How do I debug a single stage without running the whole pipeline? A: Comment out other stages in STAGES = [...], load the cached output of the previous stage, and run only the stage you're debugging.

Q: My pipeline takes too long. Where to optimize? A: (1) Use merge_lookup() — it auto-optimizes single-key joins with Series.map() which is 10x faster. (2) Cache intermediate results so you don't recompute. (3) Use usecols in load_lookup_sheet() to load only needed columns.

Reference Files

For detailed examples of each pattern, refer to:

• references/config_examples.md — Config structures across multiple domains
• references/utility_functions.md — Key utility function signatures and usage
• references/pipeline_stages.md — How multi-stage pipelines flow with variants