3. 实时数据接入层设计

实时数据接入层是整条链路的第一道关口，它决定了后续 Bronze 表能否稳定、低延迟、可追溯地拿到数据。在 Databricks 上，常见的源包括云对象存储中的文件（S3、ADLS、GCS）、Kafka、Kinesis、Event Hubs 等。它们的数据形态差异很大：文件源是"批量到达的静态对象"，消息源是"持续追加的事件流"。如果为每种源单独写一套接入逻辑，Schema 演进、断点续传、背压处理都要重复实现。本节的目标是构建一个统一接入方案：用 Auto Loader 处理文件类源，用 Structured Streaming 连接器处理消息类源，并在两者之上统一 Schema 演进策略、文件通知机制和背压控制。

一、统一接入的分层结构

统一接入方案可以拆成四层：源适配层、读取层、Schema 治理层、写入层。源适配层负责把不同源的连接参数、认证方式、偏移量语义抽象成统一配置；读取层用 Auto Loader 或 Kafka/Kinesis/Event Hubs 连接器把数据读成 DataFrame 流；Schema 治理层负责推断、校验、演进和隔离脏数据；写入层把流落到 Bronze 表，并维护检查点与元数据。

下面这张图展示了统一接入层的信号链与数据流关系，重点标出了文件通知、Schema 演进和背压三个关键控制点分别作用在链路的哪个位置。

二、Auto Loader：文件类源的统一入口

Auto Loader 的核心价值在于"增量发现 + 可扩展 Schema 推断"。它有两种文件发现模式：目录列表模式（directory listing）和文件通知模式（file notification）。目录列表模式通过定期 list 对象存储前缀来发现新文件，实现简单，但在文件量大时 list 调用成本高、延迟大。文件通知模式则借助云原生事件服务——S3 配 SQS、ADLS 配 Event Grid、GCS 配 Pub/Sub——由存储侧主动推送"新文件到达"事件，Auto Loader 消费这些事件即可，无需反复扫描。

选择依据很直接：如果源目录每天新增文件在数千以内，目录列表模式足够；如果每天新增数万甚至更多，或者对延迟敏感（要求秒级发现），应启用文件通知模式。配置上，文件通知模式需要提供队列的 URL 或连接信息，例如：

spark.readStream.format("cloudFiles")
  .option("cloudFiles.useNotifications", "true")
  .option("cloudFiles.queueUrl", "https://sqs.us-east-1.amazonaws.com/123456789/auto-loader-queue")
  .option("cloudFiles.schemaLocation", "abfss://meta@datalake.dfs.core.windows.net/schema/orders")
  .option("cloudFiles.format", "json")
  .load("abfss://raw@datalake.dfs.core.windows.net/orders/")

其中 cloudFiles.schemaLocation 是 Schema 推断结果的持久化位置，必须与检查点分开存放。Auto Loader 会在该位置维护 _schemas 目录，记录当前已知的 Schema 版本。当新文件带来新列时，Auto Loader 会更新 Schema 并继续读取，而不是让整个流失败。

三、消息类源：Kafka、Kinesis、Event Hubs

消息类源用 Structured Streaming 连接器接入。Kafka 使用 kafka 格式，Kinesis 使用 kinesis 格式，Event Hubs 则通过 Kafka 兼容接口或原生连接器接入。三者的共同点是：数据以事件为单位持续追加，读取位置由 offset 或 sequence number 标识，检查点负责记录消费进度。

以 Kafka 为例，关键配置包括 kafka.bootstrap.servers、subscribe（主题名）、startingOffsets（首次启动位置，通常用 latest 或 earliest）、maxOffsetsPerTrigger（每批次最大拉取量）。Event Hubs 的 Kafka 兼容模式复用同一套参数，只需把 bootstrap servers 指向 Event Hubs 命名空间，并配置 SASL 认证。Kinesis 则使用 streamName、region、initialPosition 等参数。

消息源与文件源在 Schema 处理上有一个重要差异：消息的 value 通常是二进制或字符串，需要显式解析。常见做法是把 value 当作 JSON 字符串，用 from_json 配合预定义 schema 解析，解析失败的记录写入 _corrupt_record 列，而不是直接丢弃。

四、Schema 演进策略

Schema 演进是接入层最容易出问题的地方。生产环境中，上游系统随时可能新增字段、改变字段类型或临时缺失字段。统一策略应包含三个动作：

• 新增列自动合并：Auto Loader 设置 cloudFiles.schemaEvolutionMode 为 addNewColumns（默认），新列自动加入 Schema 并写入 Bronze 表。消息源则需要在解析层用 mergeSchema 或手动合并。
• 类型冲突隔离：当同一字段在不同文件中类型不一致时，Auto Loader 会把冲突记录放入 _rescued_data 列，保留原始内容，避免整批失败。消息源解析失败时写入 _corrupt_record。
• Schema 版本可追溯：schemaLocation 下的 _schemas 目录保留了每次演进的版本，便于回溯"哪一批数据引入了新列"。

五、背压控制

背压的本质是"读取速率不能超过下游处理能力"。如果每批次拉取过多文件或过多 offset，会导致单批次处理时间过长、小文件堆积、内存压力上升。控制手段主要有两个：

• 文件源：maxFilesPerTrigger 限制每批次最多处理多少个文件，默认 1000。文件小、数量多时可降到 100～200；文件大、数量少时可适当提高。
• 消息源：maxOffsetsPerTrigger 限制每批次最多拉取多少条消息。Kafka 和 Event Hubs 都支持该参数，Kinesis 用 maxRecordsPerFetch 控制单次拉取量。

判断背压是否合理，可以观察流查询的 numInputRows 和 batchDuration。如果 batchDuration 持续接近或超过触发间隔，说明单批次处理不过来，应下调每批次上限；如果 numInputRows 长期远低于上限，说明资源有富余，可适当上调以提升吞吐。

六、统一接入的落地检查清单

把上述配置落到一个统一的接入模板中，文件源和消息源就可以共享同一套 Bronze 写入逻辑、同一套元数据规范和同一套监控指标。这样做的直接收益是：新增一个数据源时，只需要在源适配层增加连接配置，读取、治理、写入三层的代码和运维流程完全复用，接入层的扩展成本从"每源一套"降到"每源一份配置"。