原文： Efficient Data Processing in Python: Batch vs Streaming Pipelines Explained

每条数据管道在编写任何代码之前，都要做出一个根本选择：是按照计划分批处理数据，还是在数据到达时持续处理？

这个选择——批处理还是流式处理——会决定下游所有环节的架构。你使用的工具、能对数据时效性作出的保证、错误处理的复杂程度，以及运行整套系统所需的基础设施，都直接取决于这个决定。

选错的代价很高。本来使用批处理就足够，却搭建了流式处理管道的团队，最终不得不为一个原本不需要复杂方案的问题维护复杂的基础设施。

而在实际用例需要实时处理时仍然搭建批处理管道的团队，往往会在最糟糕的时刻才发现两者之间的差距——例如利益相关者问起，为什么数据看板已经落后了六个小时。

本文将介绍批处理管道和流式处理管道究竟是什么、两者在架构和权衡方面有何不同，以及如何使用 Python 实现这两种模式。读完本文后，你将掌握一套清晰的判断框架，可以为遇到的任何数据工程问题选择合适的方案。

前置知识

为了顺利理解本文内容，请确保你具备以下基础：

  • 有编写 Python 函数和使用模块的实践经验
  • 熟悉 pandas DataFrame 和基本的数据操作
  • 大致了解 ETL 管道的作用——抽取（extract）、转换（transform）和加载（load）

目录

  • 前置知识
  • 什么是批处理管道
    • 使用 Python 实现批处理管道
    • 批处理适合哪些场景
  • 什么是流式处理管道
    • 使用 Python 实现流式处理管道
    • 流式处理适合哪些场景
  • 关键区别一览
  • 如何在批处理与流式处理之间做出选择
  • 混合模式：Lambda 架构与 Kappa 架构

什么是批处理管道

批处理管道会将一组有明确边界且数量有限的记录作为一个整体进行处理，例如一个文件、一个数据库快照，或者一天内产生的交易记录。它会按照计划定期运行，例如每小时、每晚或每周一次，读取相应时间段内的全部数据，对数据进行转换，再将结果写入某个位置。之后，管道会停止运行，等待下一次启动。

这种模式的思路很简单：先收集，再处理。两次运行之间不会发生任何处理。

以零售行业的 ETL 为例，一条典型的批处理管道可能采用以下流程：

  1. 在午夜从事务数据库中提取过去 24 小时内产生的所有订单
  2. 将订单数据与产品目录表和客户维度表连接
  3. 按地区和产品类别计算每日营收汇总数据
  4. 将结果加载到数据仓库中，供报表使用

管道启动、完成处理，并生成一份完整、一致的昨日业务快照。分析师早晨开始工作时，数据仓库已经完成更新。

使用 Python 实现批处理管道

最简单的批处理管道可以是一个 Python 脚本，其中包含三个明确分离的阶段：抽取、转换和加载。

import pandas as pd
from datetime import datetime, timedelta

def extract(filepath: str) -> pd.DataFrame:
    """从每日导出的文件中加载原始订单数据。"""
    df = pd.read_csv(filepath, parse_dates=["order_timestamp"])
    return df

def transform(df: pd.DataFrame) -> pd.DataFrame:
    """清洗订单数据，并按地区汇总每日营收。"""
    # 只保留已完成的订单
    df = df[df["status"] == "completed"].copy()

    # 从时间戳中提取日期，以便进行分组
    df["order_date"] = df["order_timestamp"].dt.date

    # 聚合：计算每个地区每天的总营收、订单数量和平均订单金额
    summary = (
        df.groupby(["order_date", "region"])
        .agg(
            total_revenue=("order_value_gbp", "sum"),
            order_count=("order_id", "count"),
            avg_order_value=("order_value_gbp", "mean"),
        )
        .reset_index()
    )
    return summary

def load(df: pd.DataFrame, output_path: str) -> None:
    """将汇总结果写入数据仓库（此处写入 CSV 文件）。"""
    df.to_csv(output_path, index=False)
    print(f"已将 {len(df)} 行数据写入 {output_path}")

# 运行管道
raw = extract("orders_2024_06_01.csv")
aggregated = transform(raw)
load(aggregated, "warehouse/daily_revenue_2024_06_01.csv")

下面逐步看看这段代码做了什么：

  • extract 读取一个表示每日订单导出数据的 CSV 文件。parse_dates 参数会让 pandas 将 order_timestamp 列解析为日期时间对象，而不是普通字符串——这一点对于转换阶段提取日期非常重要。
  • transform 完成两项工作：首先过滤掉所有未完成的订单，例如退货和取消的订单；然后按照日期和地区对剩余订单进行分组，生成营收汇总数据。.agg() 调用会在一次处理中为每个分组计算三项指标。
  • load 将结果写入目标位置。在生产环境中，这一步可能是向数据库插入数据或将数据上传到云存储，但无论具体目标是什么，所采用的模式都相同。

这三个函数被有意分开。通过将抽取、转换和加载相互分离，每个阶段都可以独立测试、替换和调试。如果转换逻辑发生变化，你不需要修改抽取或加载代码。

批处理适合哪些场景

在以下情况下，批处理管道通常是合适的选择：

  • 数据时效性要求以小时而不是秒来衡量。 每日销售报告不需要每分钟更新一次，每周运行的营销归因模型更不需要。
  • 你需要处理大规模历史数据集。 将两年的交易历史数据回填到新的数据仓库，本质上就是一项批处理作业——数据已经存在、边界明确，而你希望在一次运行中尽可能高效地完成处理。
  • 一致性比低延迟更重要。 批处理管道会生成完整的时间点快照，输出中的每一行都基于相同的输入状态计算得出。这种一致性对于财务报告、法规合规，以及任何需要稳定、可复现数据集的下游流程都很有价值。

什么是流式处理管道

流式处理管道会在数据到达时持续处理数据，可以逐条处理记录，也可以采用较小的微批次。数据集不存在所谓的“终点”——管道会持续运行，不断从消息队列、Kafka 主题或 webhook 等来源消费事件，并在每个事件到达时立即进行处理。

这种模式的思路是：边收集，边处理。管道始终处于运行状态。

仍以零售行业的 ETL 为例，流式处理管道可以在订单创建时处理相应事件：

  1. 用户在网站上下单后，系统将一个事件发布到消息队列
  2. 流式处理管道在几毫秒内消费该事件
  3. 管道对事件进行验证、扩充，并将其路由到下游系统
  4. 欺诈检测服务、库存系统和实时数据看板都会立即收到更新后的信息

这与批处理存在根本区别：数据并不是静静地存放在文件中等待处理，而是在持续流动，管道必须跟上数据流动的速度。

使用 Python 实现流式处理管道

Python 的生成器函数天然适合用作流式处理管道的构建模块。生成器每次只产生一个值，并在两次 yield 之间暂停执行——这正好对应了数据到达时逐条处理记录、而不将全部数据加载到内存中的思路。

import json
import time
from typing import Generator, Dict

def event_source(filepath: str) -> Generator[Dict, None, None]:
    """
    从文件中模拟订单事件流。
    在生产环境中，这里会从 Kafka 或消息队列消费事件。
    """
    with open(filepath, "r") as f:
        for line in f:
            event = json.loads(line.strip())
            yield event
            time.sleep(0.01)  # 模拟事件到达之间的延迟

def validate(event: Dict) -> bool:
    """检查事件是否包含必需字段，以及字段值是否有效。"""
    required_fields = ["order_id", "customer_id", "order_value_gbp", "region"]
    if not all(field in event for field in required_fields):
        return False
    if event["order_value_gbp"] <= 0:
        return False
    return True

def enrich(event: Dict) -> Dict:
    """在将事件路由到下游之前，为其添加派生字段。"""
    event["processed_at"] = time.strftime("%Y-%m-%dT%H:%M:%S")
    event["value_tier"] = (
        "high"   if event["order_value_gbp"] >= 500
        else "mid"    if event["order_value_gbp"] >= 100
        else "low"
    )
    return event

def run_streaming_pipeline(source_file: str) -> None:
    """在每个事件从数据源到达时对其进行处理。"""
    processed = 0
    skipped = 0

    for raw_event in event_source(source_file):
        if not validate(raw_event):
            skipped += 1
            continue

        enriched_event = enrich(raw_event)

        # 在生产环境中：发布到下游主题，或写入数据接收端
        print(f"[{enriched_event['processed_at']}] "
              f"订单 {enriched_event['order_id']} | "
              f"£{enriched_event['order_value_gbp']:.2f} | "
              f"等级={enriched_event['value_tier']}")
        processed += 1

    print(f"\n完成。已处理：{processed} | 已跳过：{skipped}")

run_streaming_pipeline("order_events.jsonl")

下面看看代码的具体工作方式：

  • event_source 是一个生成器函数——注意它使用的是 yield 关键字，而不是 return。每次执行到 yield event 时，函数都会暂停，并将一个事件交给调用方。管道处理完这个事件后，生成器才会继续运行并获取下一个事件。这意味着，无论数据流有多大，内存中一次都只会保留一个事件。time.sleep(0.01) 用于模拟现实中事件从消息队列陆续到达时的间隔。
  • validate 会先检查每个事件是否包含必需字段，以及字段值是否有效，然后才进行其他操作。在流式处理场景中，异常事件非常常见——网络问题、上游系统缺陷和数据模式变更都可能产生格式错误的记录。尽早验证并跳过无效事件，要比让它们继续传播到下游系统安全得多。
  • enrich 为事件添加派生字段。这里添加的是处理时间戳和订单金额等级。在生产环境中，这一步还可能与查询表连接、调用外部 API，或者应用模型预测结果。
  • run_streaming_pipeline 将各个部分组合起来。针对 event_source 的 for 循环会逐个消费事件，让每个事件依次经过 validate → enrich → route（验证 → 扩充 → 路由）阶段，并持续记录已处理和已跳过的事件数量。

流式处理适合哪些场景

在以下情况下，流式处理管道通常是合适的选择：

  • 数据时效性以秒或毫秒来衡量。 欺诈检测、实时库存更新、实时数据看板和告警系统都要求数据立即得到处理——如果批处理作业每小时才运行一次，这些系统就会失去作用。
  • 数据量大到不适合先积累再处理。 高频物联网传感器数据、点击流事件和金融行情数据每小时可能产生数百万条记录。先积累所有数据再处理通常并不现实——这需要巨大的存储空间，而且处理作业耗时过长，等结果出来时可能已经失去价值。
  • 你需要作出反应，而不只是生成报告。 流式处理管道可以针对单个事件触发下游操作，例如发送通知、阻止交易或更新推荐结果。批处理管道只能报告已经发生的事情。

关键区别一览

下面汇总了前文讨论的批处理与流式处理之间的区别：

维度 批处理 流式处理
数据模型 有边界、有限的数据集 无边界、持续的数据流
处理触发方式 计划任务（时间或事件） 每条记录到达时触发
延迟 几分钟到几小时 几毫秒到几秒
吞吐量 高（针对批量处理进行了优化） 单条记录的开销较低
复杂度 较低 较高
状态管理 每次运行通常无状态 通常需要跨事件维护状态
错误处理 重试整个作业 针对单个事件使用死信队列
一致性 强一致性（时间点快照） 最终一致性
最适合的场景 报表、机器学习训练、数据回填 告警、实时功能、事件路由

如何在批处理与流式处理之间做出选择

这些信息固然很有用，但究竟应该如何在批处理和流式处理之间做出选择？这个决定主要取决于三个问题：

数据需要保持多高的时效性？ 如果利益相关者可以接受几小时前的结果，批处理更简单，也更具成本效益。如果他们需要在几秒内获得结果，那么流式处理就不可避免。

处理逻辑有多复杂？ 批处理作业可以连接大型数据集、执行计算成本较高的聚合，并应用复杂的业务逻辑，而不必担心延迟。流式处理管道必须快速处理每个事件，因此会限制单条记录能够承载的处理工作量。

团队具备多强的运维能力？ 流式处理基础设施——Kafka 集群、Flink 或 Spark Streaming 作业、死信队列，以及恰好一次（exactly-once）交付保证——其运维复杂度远高于按计划运行的 Python 脚本。如果团队规模较小，或者用例并不要求实时结果，那么这种复杂性只会增加成本，而不会带来实际收益。

从批处理开始。它更容易构建、测试、调试和维护。只有在某项具体、明确的需求——而不是对未来的假设——使批处理无法满足要求时，再转向流式处理。大多数数据问题都属于批处理问题，而真正需要流式处理的问题，通常会在你遇到时表现得非常明显。

正如你可能已经猜到的那样，有些数据处理系统需要将两种方式结合起来，这正是混合方案存在的原因。

混合模式：Lambda 架构与 Kappa 架构

在实践中，许多生产环境中的数据系统会同时使用这两种模式。最常见的两种混合架构是 Lambda 架构和 Kappa 架构。

Lambda 架构会并行运行一个批处理层和一个流式处理层。批处理层负责处理完整的历史数据，并在一定延迟后生成准确、一致的结果。流式处理层负责处理实时数据，并立即生成近似结果。下游消费者会合并两层的输出——利用流式处理结果保证时效性，利用批处理结果保证正确性。

这种架构的代价是运维复杂度：你需要维护两套独立的处理代码，并确保它们产生在语义上等价的结果。

Kappa 架构通过只使用流式处理层来简化系统，但当你需要进行类似批处理的重新处理时，它允许通过同一条管道重放历史数据。当 Apache Kafka 和 Apache Flink 等流式处理框架支持日志保留和重放时，这种方案会很有效。你只需要维护一套代码和一套处理逻辑，并且可以在管道发生变化后重新处理历史数据。

这两种架构都不是在所有场景下更优。Lambda 架构更常见于那些先采用批处理、后来逐步加入流式处理的组织。Kappa 架构则更常见于从一开始就将流式处理作为主要模式来设计的系统。

总结

批处理和流式处理是具有不同权衡的工具，各自适合不同类型的问题。批处理管道在一致性、简单性和批量吞吐方面表现出色；流式处理管道则擅长实现低延迟、事件响应和持续处理。

在选择 Apache Spark、Kafka 或 Flink 等具体框架之前，先从架构层面理解这两种模式，可以帮助你判断哪一种更适合当前问题，并清楚地解释这一选择。框架负责实现这些模式，而哪种模式适合你的问题，则需要由你先作出判断。