3. 数据接入与统一存储层建设

在 Databricks 机器学习平台上，统一存储层承担着“所有特征、样本、标签、日志与中间结果的唯一可信来源”这一角色。如果这一层设计不当，后续的特征工程、模型训练和在线推理都会反复陷入口径不一致、回溯困难、重复计算的问题。本节围绕 Delta Lake 表设计、数据摄取模式、批流一体接入与数据质量校验四个相互咬合的环节展开，重点说明“怎么落地”，而不是停留在概念层面。

一、Delta Lake 表设计：从访问模式反推物理结构

Delta Lake 表设计的第一步不是选字段，而是明确访问模式。机器学习场景中常见的访问模式有三类：按时间窗口回溯训练样本、按实体键（如 user_id、item_id）点查特征、按批次全量重算。这三类访问模式决定了分区键、Z-Order 列和文件布局。

分区键应选择低基数且查询中高频过滤的列，例如 dt（日期）。如果直接按 user_id 分区，会产生海量小文件，反而拖慢查询。对于需要按实体键点查的表，应使用 OPTIMIZE ... ZORDER BY (user_id) 将相关数据聚集到同一批文件中，使点查能跳过大量文件。下面是一个特征表的建表示例：

CREATE TABLE IF NOT EXISTS ml.feature_user_profile (
  user_id        STRING,
  dt             DATE,
  age_bucket     INT,
  city_level     STRING,
  last_active_ts TIMESTAMP,
  order_cnt_30d  BIGINT,
  gmv_30d        DECIMAL(18,2),
  updated_at     TIMESTAMP
)
USING DELTA
PARTITIONED BY (dt)
TBLPROPERTIES (
  'delta.autoOptimize.optimizeWrite' = 'true',
  'delta.autoOptimize.autoCompact'   = 'true',
  'delta.enableChangeDataFeed'       = 'true'
);

其中 delta.enableChangeDataFeed 对机器学习尤为重要：它让下游可以按版本读取增量变化，用于特征回填和样本对齐。表设计还应明确“主键 + 时间戳”的语义，通常用 MERGE INTO 实现幂等写入，避免重复摄取造成样本污染。

二、数据摄取模式：批、流与增量拉取的选择

摄取模式的选择取决于源系统的能力和时效要求。常见有三类：

  • 批量覆盖/追加：适用于 T+1 离线数仓表、文件落地场景，用 COPY INTO 或 INSERT OVERWRITE 完成。
  • 流式摄取：适用于 Kafka、Kinesis、事件日志，用 Structured Streaming 持续写入 Delta 表。
  • 增量拉取：适用于支持游标或更新时间戳的数据库，用 merge 按 updated_at 增量合并。

实际工程中往往混合使用。例如用户画像表每天凌晨用批量任务全量刷新，而行为事件表用流式任务持续写入。关键判断依据是：源端是否可重放、延迟容忍度、以及下游对数据新鲜度的要求。

三、批流一体接入：用同一张 Delta 表承载两种写入

批流一体的核心思想是：批和流写入同一张 Delta 表，下游读取时无需区分来源。实现要点有三：

  1. 统一 schema：批任务和流任务使用同一套字段定义，避免类型漂移。
  2. 统一分区：流式写入也按 dt 分区，保证批流数据在同一分区下可合并。
  3. 统一幂等键：流式写入使用 foreachBatch + MERGE，按业务主键去重。

下面是一个流式写入并做幂等合并的示例：

def upsert_to_delta(micro_batch_df, batch_id):
    micro_batch_df.createOrReplaceTempView("stg_events")
    spark.sql("""
      MERGE INTO ml.feature_user_profile AS t
      USING stg_events AS s
      ON t.user_id = s.user_id AND t.dt = s.dt
      WHEN MATCHED THEN UPDATE SET *
      WHEN NOT MATCHED THEN INSERT *
    """)

(spark.readStream
  .format("kafka")
  .option("subscribe", "user_events")
  .load()
  .selectExpr("CAST(value AS STRING) AS json")
  .select(from_json("json", schema).alias("d"))
  .select("d.*")
  .writeStream
  .foreachBatch(upsert_to_delta)
  .option("checkpointLocation", "/checkpoints/user_profile")
  .start())

这样，批任务和流任务写入的是同一张表，下游特征读取逻辑完全一致。

四、数据质量校验：在写入前后设置检查点

数据质量校验应嵌入摄取链路，而不是事后补救。推荐在三个位置设置检查点：

  • 写入前：校验 schema 一致性、非空约束、枚举值范围。
  • 写入后：校验行数波动、主键唯一性、分区完整性。
  • 下游读取前：校验特征分布是否漂移，例如均值、分位数是否超出阈值。

Databricks 中可用 Delta Live Tables 的 EXPECT 约束，或在普通作业中用 SQL 断言。例如：

-- 写入后校验：主键唯一
SELECT COUNT(*) AS dup_cnt
FROM (
  SELECT user_id, dt, COUNT(*) c
  FROM ml.feature_user_profile
  GROUP BY user_id, dt
  HAVING c > 1
);

若 dup_cnt > 0，则触发告警并阻断下游任务。质量校验的结果应记录到一张审计表中，便于回溯。

批流一体接入与统一存储层数据流 离线数仓 / 文件 T+1 批量数据 Kafka / 事件流 实时行为事件 业务数据库 增量拉取 updated_at 摄取与合并 COPY INTO / INSERT Structured Streaming MERGE INTO 幂等写入 统一 schema / 分区 checkpoint 管理 数据质量校验 schema / 非空 / 枚举 主键唯一 / 行数波动 分布漂移阈值 Delta Lake 统一表 ml.feature_user_profile 分区: dt Z-Order: user_id CDF 开启 autoOptimize 开启 特征读取 训练 / 推理 审计表 质量结果 校验失败阻断 数据流 质量校验流 失败阻断反馈 Delta 统一表

五、落地要点与常见陷阱

第一，不要为每类数据建一张独立表再在读取时 join，应尽量在写入阶段完成宽表化，减少训练时的 shuffle。第二，流式写入必须设置 checkpoint，否则重启后会重复消费。第三，质量校验规则要版本化，规则变更应记录在审计表中，避免“今天通过、明天失败”却无法解释。第四，Delta 表的小文件问题要持续治理，建议每天或每若干批次执行一次 OPTIMIZE，并对高频过滤列做 Z-Order。

把以上四层——表设计、摄取模式、批流一体、质量校验——串起来，统一存储层才真正成为机器学习平台的可信底座。后续的特征工程、样本构建和模型训练，都建立在这一层之上。

← 上一章 📖 返回目录 下一章 →