3. 生成式AI数据接入与摄取

生成式AI应用的效果上限，往往在数据接入阶段就已经被决定。无论是构建RAG知识库、微调多模态模型，还是驱动实时智能体，第一步都是把云存储中的文件、Kafka中的事件流、外部API返回的JSON或二进制内容，稳定、可追溯、可增量地送进湖仓。本节围绕Databricks上的两条主线展开：Auto Loader负责云存储中海量文件的增量发现与解析，Delta Live Tables（DLT，现称Lakeflow Declarative Pipelines）负责把流式与批式数据编排成有质量约束的管道。二者结合，可以覆盖从对象存储、消息队列到HTTP API的多模态摄取场景。

一、三条摄取通道的定位与差异

云存储、Kafka、API在数据形态和接入语义上差别很大，不能套用同一种写法。云存储适合“文件到达即处理”，典型对象是PDF、图片、音频、视频、JSON日志；Kafka适合“事件持续追加”，典型对象是用户行为、设备遥测、变更日志；API适合“按需拉取或轮询”，典型对象是第三方SaaS返回的JSON、分页列表、带鉴权的二进制资源。三者的关键差异在于：云存储靠目录扫描与文件通知发现新数据，Kafka靠offset维护消费位置，API靠游标、时间戳或分页token维护拉取进度。理解这一点，才能正确选择Auto Loader、Kafka连接器或自定义轮询逻辑。

二、Auto Loader：云存储增量摄取的核心机制

Auto Loader 通过 cloudFiles 数据源工作，它有两种文件发现模式：目录列表模式和文件通知模式。目录列表模式无需额外云资源，适合中小规模；文件通知模式依赖云厂商的事件通知（如 S3 EventBridge、ADLS Event Grid），适合海量文件、低延迟场景。无论哪种模式，Auto Loader 都会维护一个 RocksDB 状态存储，记录已处理文件，从而保证“新文件只处理一次”。

一个典型的文本摄取示例：

raw_text = (spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "text")
  .option("cloudFiles.schemaLocation", "s3://lake/_schemas/text/")
  .option("cloudFiles.inferColumnTypes", "true")
  .load("s3://lake/raw/docs/"))

(raw_text.writeStream
  .option("checkpointLocation", "s3://lake/_chk/text/")
  .trigger(availableNow=True)
  .toTable("bronze.docs_text"))

这里 schemaLocation 用于存放推断出的 schema 及演进历史，checkpointLocation 用于记录流进度。trigger(availableNow=True) 表示“把当前积压一次性处理完再停止”，非常适合按小时或按天调度的准实时场景；若需要持续低延迟，则改用默认的微批触发。对于图片、音频等二进制文件，Auto Loader 可以先用 binaryFile 格式读入，得到 path、content（二进制）、length 等字段，再在后续步骤中调用解析函数或模型服务提取特征。

三、Kafka 流式摄取：offset 与恰好一次

Kafka 接入使用 read_kafka，核心是把 broker、topic、消费组和起始位置配置清楚。示例：

events = (spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "broker1:9092,broker2:9092")
  .option("subscribe", "user_events")
  .option("startingOffsets", "latest")
  .option("kafka.security.protocol", "SASL_SSL")
  .load())

parsed = (events
  .selectExpr("CAST(value AS STRING) AS json_str", "timestamp")
  .select(from_json("json_str", schema).alias("d"), "timestamp")
  .select("d.*", "timestamp"))

Kafka 消息的 value 通常是字节流，需要先转成字符串再解析 JSON。offset 由 checkpoint 维护，配合 Delta 的幂等写入，可以实现端到端的恰好一次语义。startingOffsets 只在首次启动时生效，之后由 checkpoint 接管；如果误删 checkpoint，流会按 startingOffsets 重新开始，可能造成重复，因此 checkpoint 目录必须与表一起纳入备份和权限管理。

四、API 摄取：轮询、落地、再增量

API 没有原生流式连接器，常见做法是“定时轮询 + 落地文件 + Auto Loader 增量读取”。轮询任务用 Databricks Job 或 Lakeflow 调度，每次携带上次的游标或时间戳，把返回的 JSON 写入云存储的 /raw/api/ 目录，文件名带时间戳保证唯一。随后 Auto Loader 自动发现这些新文件并写入铜层。这样做的价值在于：API 的鉴权、分页、限流逻辑与湖仓摄取解耦，失败重试只影响轮询任务，不会污染流状态。

五、用 DLT 把三条通道编排成可治理管道

Delta Live Tables 用声明式方式定义表和流，并自动管理依赖、重试、血缘和质量约束。下面把云存储、Kafka 两条通道汇入同一套铜银层：

import dlt
from pyspark.sql.functions import col, from_json

@dlt.table(name="bronze_docs")
def bronze_docs():
    return (spark.readStream.format("cloudFiles")
      .option("cloudFiles.format", "text")
      .option("cloudFiles.schemaLocation", "s3://lake/_schemas/docs/")
      .load("s3://lake/raw/docs/"))

@dlt.table(name="bronze_events")
def bronze_events():
    return (spark.readStream.format("kafka")
      .option("kafka.bootstrap.servers", "broker1:9092")
      .option("subscribe", "user_events")
      .load())

@dlt.table(name="silver_docs")
@dlt.expect_or_drop("non_empty", "length(value) > 0")
def silver_docs():
    return dlt.read_stream("bronze_docs").select(
        col("value").alias("content"), col("_metadata.file_path").alias("src"))

@dlt.expect_or_drop 表示不满足约束的行直接丢弃并计入指标；若希望保留但标记，可用 expect；若希望整条管道失败，用 expect_or_fail。DLT 会在事件日志中记录每条期望的通过/失败数量，这是判断数据质量是否达标的第一手依据。多模态场景下，可以在银层调用 ai_query 或自定义 UDF，对图片做OCR、对音频做转写、对PDF做版面解析，把非结构化内容转成可检索的文本与向量。

六、结果判断与常见排障

摄取是否健康，可以从四个信号判断：一是铜层表行数是否随新文件/新消息持续增长；二是 _rescued_data 是否出现非空比例异常升高，说明上游格式漂移；三是 DLT 事件日志中期望约束的失败率是否超过阈值；四是 checkpoint 目录是否持续更新。若流停止推进，优先检查 checkpoint 路径权限、schema 冲突和 Kafka 消费组是否被其他作业占用。对于 API 轮询，重点看游标是否被正确持久化，避免每次全量拉取造成重复。

把三条通道统一到铜银层之后，生成式AI应用就可以在银层之上构建向量索引、特征表或检索服务。摄取阶段留下的文件路径、摄取时间、质量指标，正是后续做数据溯源、增量刷新和问题定位的关键依据。