2. 配置驱动的数据管道

任务说明

构建一个配置驱动的数据管道：管道的行为（数据源、转换步骤、输出目标）完全由外部JSON配置文件决定，代码本身不包含任何业务数据路径的硬编码。修改管道行为时只需编辑配置文件，无需改动Python代码。

实现方案

采用“注册表+分发器”模式：定义转换函数的注册表，管道引擎读取配置后，根据配置中的名称动态调用对应函数。数据源与输出目标通过配置中的类型字段分发到对应的读写处理器。整个管道按“读取→依次转换→写出”的顺序执行，每一步的输入输出均为字典列表（表格型数据）。

代码实现

import csv
import json
from pathlib import Path

# ---------- 转换函数注册表 ----------
TRANSFORMS = {}

def register(name):
    """装饰器：将函数注册到转换注册表"""
    def decorator(func):
        TRANSFORMS[name] = func
        return func
    return decorator

@register("filter_rows")
def filter_rows(rows, field, value):
    """保留指定字段等于给定值的行"""
    return [r for r in rows if r.get(field) == value]

@register("add_column")
def add_column(rows, field, value):
    """为每行添加一个常量字段"""
    for r in rows:
        r[field] = value
    return rows

@register("drop_fields")
def drop_fields(rows, fields):
    """删除指定字段列表"""
    drop_set = set(fields)
    result = []
    for r in rows:
        result.append({k: v for k, v in r.items() if k not in drop_set})
    return result

# ---------- 数据源与输出目标 ----------
def read_csv(path):
    with open(path, newline='', encoding='utf-8') as f:
        return list(csv.DictReader(f))

def write_json(rows, path):
    with open(path, 'w', encoding='utf-8') as f:
        json.dump(rows, f, ensure_ascii=False, indent=2)

def read_json(path):
    with open(path, encoding='utf-8') as f:
        return json.load(f)

def write_csv(rows, path):
    if not rows:
        Path(path).write_text('', encoding='utf-8')
        return
    fieldnames = list(rows[0].keys())
    with open(path, 'w', newline='', encoding='utf-8') as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(rows)

# ---------- 管道引擎 ----------
def run_pipeline(config_path):
    """根据配置文件执行数据管道"""
    with open(config_path, encoding='utf-8') as f:
        config = json.load(f)

    # 1. 读取数据源
    src = config["source"]
    if src["type"] == "csv":
        rows = read_csv(src["path"])
    elif src["type"] == "json":
        rows = read_json(src["path"])
    else:
        raise ValueError(f"不支持的数据源类型: {src['type']}")

    # 2. 依次应用转换
    for step in config.get("transforms", []):
        name = step["name"]
        if name not in TRANSFORMS:
            raise KeyError(f"未注册的转换: {name}")
        params = {k: v for k, v in step.items() if k != "name"}
        rows = TRANSFORMS[name](rows, **params)

    # 3. 写出到目标
    sink = config["sink"]
    if sink["type"] == "json":
        write_json(rows, sink["path"])
    elif sink["type"] == "csv":
        write_csv(rows, sink["path"])
    else:
        raise ValueError(f"不支持的输出类型: {sink['type']}")

    return rows

# ---------- 主程序入口 ----------
if __name__ == "__main__":
    import sys
    config_file = sys.argv[1] if len(sys.argv) > 1 else "pipeline.json"
    result = run_pipeline(config_file)
    print(f"管道执行完成，共 {len(result)} 行数据")

配置文件示例

创建 pipeline.json，定义数据源为CSV、依次执行两个转换、输出为JSON：

{
  "source": {
    "type": "csv",
    "path": "input.csv"
  },
  "transforms": [
    { "name": "filter_rows", "field": "department", "value": "Engineering" },
    { "name": "add_column", "field": "source", "value": "internal" }
  ],
  "sink": {
    "type": "json",
    "path": "output.json"
  }
}

输入数据示例

创建 input.csv：

name,department,salary
Alice,Engineering,85000
Bob,Sales,72000
Carol,Engineering,91000

运行与验证

运行方式：

python pipeline.py pipeline.json

验证步骤：

1. 运行上述命令，观察终端输出“管道执行完成，共 2 行数据”。
2. 打开生成的 output.json，确认内容为：

[
  {
    "name": "Alice",
    "department": "Engineering",
    "salary": "85000",
    "source": "internal"
  },
  {
    "name": "Carol",
    "department": "Engineering",
    "salary": "91000",
    "source": "internal"
  }
]

判定标准：

• 输出文件仅包含 department 为 Engineering 的行，且每行新增了 source 字段，即管道正确执行了配置中的两个转换。
• 修改 pipeline.json 中的过滤值或增删转换步骤后重新运行，输出随之变化而代码未改动，即证明配置与代码成功分离。