管道架构总览：我眼中的整体拼图

说到管道架构，很多人一上来就纠结细节。哪个组件先跑，谁调用谁，结果写出来的代码改来改去。

我个人习惯，先画一张大图。搞清楚每个模块的位置，再下手。你想想看，盖房子总得看蓝图吧？

管道的四大核心组件

我把整个管道拆成了四个部分。每个部分职责清晰，互不干扰。这是我在项目中踩过坑后总结的——之前把判断逻辑和存储逻辑揉在一起，改一个地方动全身，难受得很。

组件	职责	输入	输出
数据接入器	加载原始批次数据，做格式校验	文件路径 / Kafka 流	标准化批次记录
校验引擎	执行预定义的各维度校验规则	标准化记录 + 规则集	校验中间结果
聚合器	合并多批次、多规则的中间结果	校验中间结果（多个）	聚合后的统计量
报告生成器	将聚合结果格式化为报告并存储	聚合统计量 + 模板	JSON/PDF 报告

你看，每个组件只需要知道它前面的一步输入，和后面一步的输出。这样就形成了清晰的接口边界。

数据流与中心调度器

组件之间怎么联动？我建议不要直接互相调用。否则，以后想加一个新组件，就得改所有上下游的代码。

一个轻量级的中心调度器就能解决这个问题。它负责四件事：

• 监控数据接入器的输出队列
• 管理校验引擎的线程池
• 触发聚合器的统计任务
• 串行化报告生成操作

我曾经在一个项目里试过用 Celery 做调度。后来发现，对于批次数百级别的场景，自己写一个 Python 异步事件循环反而更可控。调度逻辑就 80 行代码，出了问题自己就能修。

SVG：管道架构总览

下面这张图是我在笔记本上画过很多次的版本。它把上面说的所有东西串起来了。

各组件的协作接口

接口是管道能否灵活扩展的关键。我一般会为每个组件定义一个基础抽象类。这样后续替换实现，比如从本地文件读取换成从 S3 读取，只需要继承基类即可。

class BaseDataLoader(ABC):
    @abstractmethod
    def load(self, source: str) -> Iterator[BatchRecord]:
        pass

class BaseValidator(ABC):
    @abstractmethod
    def validate(self, record: BatchRecord, rules: list) -> ValidationResult:
        pass

class BaseAggregator(ABC):
    @abstractmethod
    def aggregate(self, results: list[ValidationResult]) -> AggregatedStats:
        pass

class BaseReporter(ABC):
    @abstractmethod
    def generate(self, stats: AggregatedStats) -> Report:
        pass

你看，每个接口的输入输出都是明确的类型。这样做的好处是，单元测试的时候，你随便 mock 一个组件，不影响其他部分的逻辑验证。

校验管道与训练循环的关系

很多人会问，这个管道是独立的，还是嵌入到训练流程中的？嗯，这个得看场景。

• 离线模式：训练结束后，读取保存的批次 log，一次性跑校验。适合模型评估阶段。
• 在线模式：每个训练批次结束后，异步触发校验。适合监控训练质量。

我个人更推荐离线模式起步。为什么呢？因为在线模式会引入「数据竞争」问题——你校验引擎还没读完上一批，训练器已经把下一批写进来了。虽然可以用锁解决，但复杂度一下子就上去了。

管道配置示例

为了让管道可配置，我习惯把所有参数写到一个 YAML 文件里。这样改批次大小、改校验规则，都不用改代码。

pipeline:
  loader:
    type: local_file
    path: ./data/batch_logs/
    format: jsonl
    batch_size: 128

  validator:
    rules:
      - name: schema_check
        enabled: true
      - name: value_range
        enabled: true
        params:
          thresholds: [-10.0, 10.0]
      - name: consistency_check
        enabled: false

  aggregator:
    statistics:
      - mean
      - std
      - missing_rate

  reporter:
    output_dir: ./reports/
    template: standard

为什么这样设计？你想想看，如果换了项目，只要改这个配置文件就行。代码完全复用。我在一个三年前的项目里就是这么做的，后来换了数据集，两个小时就完成了管道适配。

容错与重试机制

管道跑起来，难免会遇到各种问题。网络抖动、磁盘满、批次数据损坏。这时候不能直接崩掉，得有容错。

class RetryHandler:
    def __init__(self, max_retries=3, backoff=1.0):
        self.max_retries = max_retries
        self.backoff = backoff

    def run_with_retry(self, func, *args, **kwargs):
        for attempt in range(self.max_retries):
            try:
                return func(*args, **kwargs)
            except DataFormatError:
                # 数据格式错误，重试也没用，直接抛弃
                raise
            except ConnectionError as e:
                if attempt == self.max_retries - 1:
                    raise TimeoutError("重试耗尽") from e
                time.sleep(self.backoff * (2 ** attempt))
            except Exception as e:
                # 记录异常批次ID，稍后人工排查
                logging.error(f"Unexpected error: {e}")
                raise

注意这里的分类处理。数据格式错误，重试一万次也没用，直接跳过。连接超时，可以等一会再试。我曾经遇到过一个案例，每次跑第三个批次就挂，后来发现是临时文件没清理干净。加了日志后，这种问题十分钟就能定位。