📚 课程目录第1章批流一体数据处理概览

1. 批流一体数据处理概览

在传统数据平台里，批处理与流处理往往被拆成两套系统：一套负责每天或每小时跑一次的大规模历史数据计算，另一套负责秒级、毫秒级到达数据的实时计算。两套系统各有各的存储、计算引擎、元数据、权限和运维方式，业务想要“同一份口径既看历史又看实时”，通常要在中间做大量数据搬运和逻辑对齐。批流一体的核心目标，就是让这两类负载在同一套存储、同一套元数据、同一套计算引擎和同一套权限体系下协同工作，而不是简单地把两个集群拼在一起。

批处理与流处理的本质差异

批处理面对的是“有界数据集”：数据已经落在文件或表中，作业启动时能知道数据边界，可以按分区、按文件并行扫描，适合全量重算、复杂关联、历史回溯。它的典型特征是吞吐高、延迟以分钟到小时计、失败后可以整批重跑。流处理面对的是“无界数据流”：数据持续到达，没有明确的结束点，计算必须随着事件不断推进，输出延迟通常在秒级甚至更低。它更关注事件时间、水位线、状态管理和恰好一次语义。

两者真正的难点不在“快慢”，而在语义对齐。例如同一笔订单，批处理按天统计时把它算在“下单日”，流处理如果按到达时间统计，可能因为网络延迟落到第二天，导致实时看板和离线报表对不上。因此批流一体必须解决三件事：统一的数据存储格式、统一的计算语义（尤其是时间与状态）、统一的元数据与权限。

融合价值：为什么不是“两个系统各跑各的”

批流一体的价值可以落到三个具体层面。第一是口径一致：同一份 Delta 表既可以被批作业全量扫描，也可以被流作业增量读取，业务指标不再出现“实时看板 100 万、离线报表 98 万”的尴尬。第二是成本与运维收敛：不需要为流处理单独维护一套存储和元数据，权限、血缘、监控都复用同一套体系。第三是开发效率：同一段 SQL 或 DataFrame 逻辑，改一个 Trigger 就能从批模式切到流模式，增量逻辑不必重写。

但要注意，批流一体不等于“所有场景都用流”。历史全量重算、复杂多表关联、机器学习特征回填，仍然更适合批处理；只有对延迟敏感、需要持续更新的场景才值得上流处理。判断标准是：数据是否无界、业务是否要求秒级可见、状态是否需要长期维护。

Databricks Lakehouse 如何统一批流

Databricks 的统一路径可以概括为“一层存储、一套引擎、一份元数据”。存储层是 Delta Lake，它在对象存储之上提供 ACID 事务和事务日志，使一张表既能被批读也能被流读；流作业通过 readStream 增量消费 Delta 表的新增文件，批作业通过 read 扫描全表，两者看到的是同一份数据。计算层是 Spark，批用 spark.read + write，流用 spark.readStream + writeStream，API 结构高度对称。元数据层是 Unity Catalog，表、权限、血缘对批流一视同仁。

下面是一个最小可运行的示例：把 Kafka 中的订单事件流写入 Delta 表，再用批查询验证结果。示例参数为教学假设，实际运行时需替换为真实连接信息。

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, to_timestamp

spark = SparkSession.builder.appName("batch-stream-unified").getOrCreate()

# 1. 流式读取 Kafka 订单事件
orders_stream = (
    spark.readStream
         .format("kafka")
         .option("kafka.bootstrap.servers", "broker:9092")
         .option("subscribe", "orders")
         .option("startingOffsets", "latest")
         .load()
)

# 2. 解析 JSON，统一事件时间字段
parsed = (
    orders_stream
      .selectExpr("CAST(value AS STRING) AS json_str")
      .selectExpr("from_json(json_str, 'order_id STRING, amount DOUBLE, event_time STRING') AS d")
      .select(col("d.order_id"), col("d.amount"),
              to_timestamp(col("d.event_time")).alias("event_time"))
)

# 3. 写入 Delta 表，设置水位线容忍 10 分钟迟到数据
query = (
    parsed.writeStream
          .format("delta")
          .option("checkpointLocation", "/checkpoints/orders")
          .option("mergeSchema", "true")
          .withWatermark("event_time", "10 minutes")
          .outputMode("append")
          .start("/delta/orders")
)

# 4. 批查询验证：同一张表，批读即可
spark.read.format("delta").load("/delta/orders").createOrReplaceTempView("orders")
spark.sql("""
    SELECT date_trunc('hour', event_time) AS hour_bucket,
           COUNT(*) AS order_cnt,
           ROUND(SUM(amount), 2) AS total_amount
    FROM orders
    GROUP BY 1
    ORDER BY 1
""").show()

运行后，流作业持续把新事件追加到 /delta/orders，批查询可以随时读取同一张表做小时级汇总。判断是否成功的关键观察点有三个：Delta 表目录下是否持续产生新的 parquet 文件；_delta_log 中是否出现新的 commit；批查询结果是否随流写入而增长。如果批查询结果长时间不变，优先检查 checkpoint 路径是否可写、Kafka 是否有新消息、水位线是否把迟到数据过滤掉。

批流一体的最小验证方法：先用流写入一张 Delta 表，再用批查询读同一张表。如果两者能读到同一份数据且口径一致，说明存储与元数据已经统一；如果读不到，问题通常出在 checkpoint、权限或表路径不一致。

常见误区与判断依据

第一个误区是把“批流一体”理解成“流处理可以替代批处理”。实际上，全量重算、复杂 join、历史回溯仍然依赖批处理的高吞吐和可重跑特性。第二个误区是忽略事件时间与处理时间的区别：如果流作业用处理时间做窗口，而批作业用事件时间做分区，两者结果必然对不上。第三个误区是 checkpoint 与表路径混用：多个流作业写同一张表时，必须各自使用独立 checkpoint，否则会出现重复写入或状态错乱。

判断一个场景是否适合批流一体，可以看三个问题：数据是否持续到达且需要低延迟可见；同一份数据是否既要做实时聚合又要做历史回溯；是否希望复用同一套权限、血缘和监控。如果三个答案都是“是”，那么用 Delta Lake + Structured Streaming 统一批流就是合理选择；如果只是每天跑一次报表，批处理单独使用反而更简单、更省资源。

批流一体不是“所有作业都改成流”。流作业会长期占用计算资源并维护状态，成本通常高于批作业。对延迟不敏感的场景，优先用批处理；只有真正需要持续更新的场景，才引入流处理。