📚 课程目录第3章数据接入与统一存储

3. 数据接入与统一存储

在 Databricks 平台上，数据接入不是“把文件拖进来”这么简单，它决定了后续特征工程、模型训练和推理服务能否稳定复现。本节围绕三条主线展开：从哪里读（云存储、关系型数据库、流式源）、用什么格式存（Delta Lake、Parquet、CSV、JSON）、如何治理（注册到 Unity Catalog）。理解这三者的配合关系，比记住某个 API 更重要。

一、三类数据源与接入方式

云存储是最常见的落地形态，包括 AWS S3、Azure Data Lake Storage Gen2、Google Cloud Storage。Databricks 通过挂载点（mount）或直接使用 abfss://、s3a:// 等 URI 访问。推荐优先使用 Unity Catalog 的外部位置（External Location）授权，而不是长期依赖挂载点，因为前者能统一管理凭证与访问控制。

关系型数据库（如 MySQL、PostgreSQL、SQL Server、Oracle）通常通过 JDBC 接入。对于一次性或小批量抽取，可以用 spark.read.jdbc()；对于需要增量同步的场景，更推荐 Lakeflow Connect 或基于 COPY INTO 的批量加载，避免全表扫描带来的压力。关键参数包括 url、dbtable、user、password、numPartitions、partitionColumn、lowerBound、upperBound，其中后四项决定并行度与数据倾斜。

流式源包括 Kafka、Kinesis、Event Hubs、Auto Loader 监听的云存储目录。结构化流（Structured Streaming）用 spark.readStream 读取，配合 writeStream 写入 Delta 表。Auto Loader（cloudFiles）是 Databricks 推荐的增量文件接入方式，它能自动发现新文件、推断 Schema 并支持 Schema 演进。

二、格式选择：Delta Lake 优先，Parquet/CSV/JSON 各司其职

Delta Lake 是 Databricks 上的首选目标格式。它在 Parquet 之上增加了事务日志（_delta_log），提供 ACID 事务、时间旅行、Schema 演进、MERGE 更新和 OPTIMIZE 小文件合并。对于需要频繁更新、需要审计历史、需要多用户并发写入的场景，Delta 几乎是默认答案。

Parquet 适合只读分析场景，压缩比高、列裁剪效果好，但不支持原地更新。CSV 和 JSON 更适合作为接入过渡格式：CSV 结构简单但无类型信息、无嵌套；JSON 支持嵌套但解析开销大、Schema 易漂移。实践中常见做法是先用 Auto Loader 把 CSV/JSON 读入，做清洗和类型转换后写入 Delta 表，而不是让下游直接消费原始 CSV。

一个容易忽略的点是：格式选择会影响后续 Unity Catalog 的注册方式。Delta 表可以直接 CREATE TABLE 指向路径并自动识别 Schema；Parquet 表需要显式指定 Schema 或依赖推断；CSV/JSON 通常建议先落地为 Delta 再注册，避免每次查询都重新解析。

三、注册到 Unity Catalog 的实操路径

Unity Catalog 采用三级命名空间：catalog.schema.table。注册前需要先配置 Storage Credential 和 External Location，让 UC 有权限访问底层云存储。以下是一个从云存储读取 Parquet 并注册为 Delta 表的示例：

-- 1. 创建外部位置（由管理员执行一次）
CREATE EXTERNAL LOCATION IF NOT EXISTS ext_ml_raw
  URL 'abfss://raw@mlstore.dfs.core.windows.net/'
  WITH (STORAGE CREDENTIAL cred_mlstore);

-- 2. 创建目录与模式
CREATE CATALOG IF NOT EXISTS ml_platform;
CREATE SCHEMA IF NOT EXISTS ml_platform.bronze;

-- 3. 从 Parquet 读取并写入 Delta 表
CREATE OR REPLACE TABLE ml_platform.bronze.user_events
AS
SELECT
  user_id,
  event_type,
  CAST(event_time AS TIMESTAMP) AS event_time,
  get_json_object(properties, '$.page') AS page
FROM parquet.`abfss://raw@mlstore.dfs.core.windows.net/events/`
WHERE event_time IS NOT NULL;

-- 4. 验证注册结果
SHOW TABLES IN ml_platform.bronze;
DESCRIBE EXTENDED ml_platform.bronze.user_events;

执行后，DESCRIBE EXTENDED 会返回表的存储位置、Provider（delta）、创建时间等信息。如果 Provider 显示为 delta，说明注册成功；如果显示 parquet，则说明只是外部表映射，没有获得 Delta 的事务能力。

四、流式接入与 Schema 演进

流式场景下，Auto Loader 的典型配置如下：

df = (spark.readStream
      .format("cloudFiles")
      .option("cloudFiles.format", "json")
      .option("cloudFiles.schemaLocation", "/checkpoints/schema")
      .option("cloudFiles.inferColumnTypes", "true")
      .load("abfss://raw@mlstore.dfs.core.windows.net/streaming/"))

(df.writeStream
   .option("checkpointLocation", "/checkpoints/user_events")
   .option("mergeSchema", "true")
   .trigger(availableNow=True)
   .toTable("ml_platform.bronze.user_events_stream"))

其中 schemaLocation 用于保存推断出的 Schema，mergeSchema 允许新字段自动合并到目标 Delta 表。判断流是否正常，可以观察 writeStream 返回的 StreamingQuery 状态，或查询 Delta 表的 DESCRIBE HISTORY 看每次微批的写入记录。

实践建议：接入层建议按“青铜—白银—黄金”分层。青铜层保留原始格式与全量字段，白银层做清洗和类型规范，黄金层面向特征与指标。Unity Catalog 的目录结构可以按 catalog.bronze、catalog.silver、catalog.gold 组织，权限也按层授予，避免下游直接访问原始数据。

常见坑：CSV 读取时若未指定 header 和 inferSchema，所有列会变成 string，后续数值计算会静默出错；JSON 的嵌套字段若未用 get_json_object 或 from_json 显式提取，容易在 Schema 演进时丢失字段。注册到 UC 前，务必先用 DESCRIBE 确认列类型与预期一致。

总结来看，数据接入与统一存储的核心不是“能读进来”，而是“读进来之后能被稳定、可治理地复用”。云存储、数据库、流式源提供输入，Delta Lake 提供可靠的目标格式，Unity Catalog 提供命名、权限与血缘。三者配合，才能让后续的特征工程和模型训练建立在可信的数据基础之上。