3. Delta Lake 表基础：创建表、写入数据、读取数据、表 Schema 管理

好，咱们正式开始动手了。前面两章聊了 Delta Lake 的来龙去脉和核心原理，说白了就是让你知道它为什么牛。但从这一章开始，我们要真刀真枪地干。

创建表、写数据、读数据、管 Schema——这四个动作，是你以后每天都会跟 Delta Lake 打交道的入口。我个人的习惯是，不管项目多急，先把这几个基础操作练到肌肉记忆。为什么？因为后面所有的高级特性，像时间旅行、Z-Order 优化、流式写入，全是在这个地基上盖楼。

3.1 创建 Delta 表：不止一种姿势

Delta Lake 的表，本质上还是 Spark 的 DataFrame 或 SQL 操作，只不过底层存储格式换成了 Delta。创建方式我归纳了三种，你根据场景选。

3.1.1 从 DataFrame 直接创建

这是最常用的方式。你有一份数据，想把它存成 Delta 表，直接写就行。

# 用 Spark 创建 DataFrame
data = [(1, "Alice", 1000), (2, "Bob", 2000), (3, "Charlie", 1500)]
columns = ["id", "name", "salary"]
df = spark.createDataFrame(data, columns)

# 保存为 Delta 表
df.write.format("delta").save("/path/to/delta-table")

嗯，这里有个细节要注意。如果你不指定 mode，默认是 error，也就是说路径已存在时会报错。我刚开始用的时候就被这个坑过——跑了个批处理，第二次运行直接挂了。

3.1.2 用 SQL 语句创建

如果你团队里分析师多，他们更习惯 SQL。Delta Lake 完全支持。

-- 创建托管表
CREATE TABLE IF NOT EXISTS employee (
  id INT,
  name STRING,
  salary DOUBLE
) USING DELTA;

-- 创建外部表，指定路径
CREATE TABLE IF NOT EXISTS employee_ext
USING DELTA
LOCATION '/path/to/delta-table';

托管表和外部表的区别，说白了就是：托管表删表时数据一起删，外部表只删元数据。我在项目中一般用外部表，因为数据是资产，元数据可以重建，数据丢了就真没了。

3.1.3 从 Parquet 表升级

这个场景我遇到过好几次。公司之前用 Parquet 存了海量历史数据，想迁移到 Delta Lake 又不想重跑。Delta Lake 提供了 convertToDelta 方法，原地升级。

# 将 Parquet 表原地转换为 Delta 表
spark.sql("CONVERT TO DELTA parquet.`/path/to/parquet-table`")

3.2 写入数据：Append、Overwrite 与 Merge

表建好了，接下来就是往里灌数据。Delta Lake 的写入模式比普通 Parquet 丰富得多，我重点讲三个最常用的。

3.2.1 Append 追加写入

日志数据、事件流数据，基本都是追加。简单直接。

new_data = [(4, "David", 2500)]
df_new = spark.createDataFrame(new_data, columns)
df_new.write.format("delta").mode("append").save("/path/to/delta-table")

3.2.2 Overwrite 覆盖写入

全量刷新场景，比如每天从源系统拉一次全量数据。

df_full.write.format("delta").mode("overwrite").save("/path/to/delta-table")

但这里有个坑：overwrite 默认替换整个表。如果你只想替换某个分区，要用 replaceWhere。

# 只覆盖 2024 年的分区
df_2024.write.format("delta") \
  .mode("overwrite") \
  .option("replaceWhere", "year = 2024") \
  .save("/path/to/delta-table")

3.2.3 Merge 合并写入（UPSERT）

这才是 Delta Lake 的王牌操作。传统 Hive 或 Parquet 要做 UPSERT，得先读全量数据，比对，再写回去，又慢又容易出错。Delta Lake 的 merge 操作是原子性的。

from delta.tables import DeltaTable

deltaTable = DeltaTable.forPath(spark, "/path/to/delta-table")

# 要更新/插入的数据
updates = spark.createDataFrame([(2, "Bob Updated", 2200)], columns)

deltaTable.alias("target") \
  .merge(updates.alias("source"), "target.id = source.id") \
  .whenMatchedUpdateAll() \
  .whenNotMatchedInsertAll() \
  .execute()

这段代码的意思是：如果 ID 匹配就更新，不匹配就插入。整个过程在一个事务里完成，不会出现中间状态。

写入模式	适用场景	事务性
Append	日志、事件流、增量数据	✅ 原子追加
Overwrite	全量刷新、分区替换	✅ 原子覆盖
Merge	CDC、维表缓慢变化、数据修正	✅ 原子合并

3.3 读取数据：快照读与流式读

读数据听起来简单，但 Delta Lake 给了你两种完全不同的读取方式。

3.3.1 快照读（Batch Read）

就是普通的批式读取，拿到表的最新快照。

df = spark.read.format("delta").load("/path/to/delta-table")
df.show()

你还可以读取历史版本，这就是时间旅行的基础。

# 读取版本 5 的数据
df_v5 = spark.read.format("delta") \
  .option("versionAsOf", 5) \
  .load("/path/to/delta-table")

# 读取某个时间点的数据
df_time = spark.read.format("delta") \
  .option("timestampAsOf", "2024-01-15 10:00:00") \
  .load("/path/to/delta-table")

3.3.2 流式读（Streaming Read）

Delta Lake 天然支持 Structured Streaming。你可以把 Delta 表当作流式数据源，持续读取新增的数据。

streaming_df = spark.readStream.format("delta") \
  .load("/path/to/delta-table")

query = streaming_df.writeStream \
  .format("console") \
  .outputMode("append") \
  .start()

流式读的核心原理是：Delta Lake 的事务日志记录了每次提交的文件列表，流式引擎只需要轮询日志，就能知道哪些是新文件。不需要像 Kafka 那样维护 offset，Delta Lake 自己管。

3.4 表 Schema 管理：灵活与约束的平衡

Schema 管理是 Delta Lake 区别于传统数据湖的一大亮点。说白了，它让你既能灵活应对 Schema 变化，又能守住数据质量的底线。

3.4.1 Schema 自动演化

你写入的数据字段和表定义不一致怎么办？传统 Parquet 直接报错。Delta Lake 可以自动演化。

# 开启 Schema 自动演化
df_with_new_col.write.format("delta") \
  .option("mergeSchema", "true") \
  .mode("append") \
  .save("/path/to/delta-table")

新字段会自动加到表 Schema 末尾，旧数据的该字段值为 null。这个功能在数据源经常加字段的场景下特别好用。我在做实时数仓时，业务方三天两头加字段，要是每次都手动改 Schema，我估计得疯。

3.4.2 Schema 强制检查

但自动演化不是万能的。有时候你希望严格校验，防止脏数据进来。

# 关闭 Schema 演化，严格匹配
df.write.format("delta") \
  .option("mergeSchema", "false") \
  .mode("append") \
  .save("/path/to/delta-table")

默认情况下，Delta Lake 是严格模式。只有当你显式设置 mergeSchema=true 时才会演化。

3.4.3 Schema 约束与校验

Delta Lake 支持在表级别添加约束，比如非空检查、值范围检查。

ALTER TABLE employee ADD CONSTRAINT salary_check CHECK (salary > 0);
ALTER TABLE employee ADD CONSTRAINT id_not_null CHECK (id IS NOT NULL);

写入数据时，如果违反约束，事务会回滚并报错。这个功能在数据质量要求高的场景下非常有用。我记得有一次，上游系统抽风，灌进来一批负数工资的数据，幸亏有约束拦住了，不然报表出来就闹笑话了。

3.5 本章知识体系总览

下面这张图把本章的核心知识点串起来了。你可以把它当作操作手册的索引。

这张图把创建、写入、读取、Schema 管理四个维度以及它们的子操作都列出来了。你以后在实际项目中，遇到任何表操作问题，都可以回到这张图上找对应的方法。