📚 课程目录第4章 Delta表基础操作

4、Delta表基础操作：创建表、写入数据、读取数据

好，咱们进入实战环节。Delta Lake 说到底，核心就是那张表。你 Spark 写得再花哨，数据落不到表里，或者读不出来，那都是白搭。

这一章，我带你亲手把 Delta 表跑起来。创建、写入、读取，三板斧。别小看这些基础操作，我在生产环境里见过太多因为建表姿势不对，后面查数据查到崩溃的案例。

核心要点：Delta 表 = 数据文件（Parquet）+ 事务日志（_delta_log）。你所有的写操作，都会先记日志，再改数据。这是它跟普通 Parquet 表最大的区别。

4.1 创建 Delta 表：两种姿势

创建 Delta 表，说白了就两条路：从 DataFrame 直接存，或者用 SQL 建表再插数据。我个人习惯用 DataFrame 做原型验证，生产环境用 SQL 多一些，方便权限管理。

4.1.1 方式一：DataFrame API 创建

这是最直接的方式。你有一个 DataFrame，调用 save 或者 saveAsTable，Delta 表就出来了。

# 从 SparkSession 创建一些示例数据
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("/tmp/delta/user_info")

# 保存为 Delta 表（托管表，元数据在 Hive Metastore）
df.write.format("delta").saveAsTable("default.user_info")

这里有个细节：save 是路径表，数据文件在哪你说了算；saveAsTable 是托管表，Spark 帮你管路径。我在项目中遇到过，有人用 saveAsTable 建了一堆表，后来要迁移数据，找文件路径找了半天。所以，生产环境我建议用路径表，路径清晰，好维护。

4.1.2 方式二：SQL 方式创建

如果你团队里分析师多，他们更习惯写 SQL。Delta Lake 支持标准 CREATE TABLE 语法，加一句 USING DELTA 就行。

-- 创建托管表
CREATE TABLE IF NOT EXISTS default.user_info (
    id INT,
    name STRING,
    salary DECIMAL(10,2)
) USING DELTA;

-- 创建外部表（指定路径）
CREATE TABLE IF NOT EXISTS default.user_info_ext (
    id INT,
    name STRING,
    salary DECIMAL(10,2)
) USING DELTA
LOCATION '/tmp/delta/user_info_ext';

嗯，这里要注意：LOCATION 路径下如果已经有 Delta 数据，建表时会自动关联上。你想想看，这其实是个很实用的特性——数据已经在那了，你只需要建个“壳”就能查。

4.2 写入数据：Append、Overwrite、Upsert

表建好了，接下来就是往里写数据。Delta 的写入模式比普通 Parquet 丰富得多。我重点讲三种：Append（追加）、Overwrite（覆盖）、Upsert（更新插入）。

4.2.1 Append 模式

最常用，就是往表里加新数据。默认就是 Append。

# 追加写入
new_data = [(4, "David", 2500)]
new_df = spark.createDataFrame(new_data, ["id", "name", "salary"])
new_df.write.format("delta").mode("append").save("/tmp/delta/user_info")

追加写入不会动旧数据。你查一下表，会发现多了个 Parquet 文件，_delta_log 里也多了一条日志。

4.2.2 Overwrite 模式

全量覆盖。这个要小心，一旦覆盖，旧数据就没了。不过 Delta 有版本回溯，你还能救回来，这个后面章节会讲。

# 覆盖写入
overwrite_data = [(1, "Alice", 3000), (2, "Bob", 2500)]
overwrite_df = spark.createDataFrame(overwrite_data, ["id", "name", "salary"])
overwrite_df.write.format("delta").mode("overwrite").save("/tmp/delta/user_info")

我曾经踩过的坑：有一次跑批任务，不小心把 Overwrite 写成了 Append，结果下游报表数据翻倍。排查了半天才发现是 mode 写错了。所以，生产环境跑 Overwrite 前，一定先确认数据量级和影响范围。

4.2.3 Upsert 模式（Merge Into）

这是 Delta Lake 的王牌功能。你可以根据条件，决定哪些行插入，哪些行更新，哪些行删除。语法就是 MERGE INTO。

-- 准备源数据
CREATE OR REPLACE TEMP VIEW updates AS
SELECT 1 AS id, 'Alice' AS name, 3500 AS salary
UNION ALL
SELECT 5 AS id, 'Eve' AS name, 1800 AS salary;

-- 执行 Merge
MERGE INTO default.user_info AS target
USING updates AS source
ON target.id = source.id
WHEN MATCHED THEN UPDATE SET target.name = source.name, target.salary = source.salary
WHEN NOT MATCHED THEN INSERT *;

这个例子做了两件事：id=1 的 Alice 工资从 3000 更新到 3500；id=5 的 Eve 不存在，直接插入。一次操作，搞定更新和插入。我在做实时数仓时，CDC 数据入湖基本全靠 Merge Into，效率比先删后插高太多了。

4.3 读取数据：批式与流式

读数据就简单多了。Delta 表读起来跟普通表一样，但多了个时间旅行（Time Travel）的能力。

4.3.1 批式读取

# 读取 Delta 表
df = spark.read.format("delta").load("/tmp/delta/user_info")
df.show()

# 读取最新版本
df_latest = spark.table("default.user_info")
df_latest.show()

没什么特别的，就是标准 Spark 读法。但 Delta 有个好处：读的时候不会锁表。你写你的，我读我的，互不干扰。这在实时场景下特别重要。

4.3.2 时间旅行读取

想查某个历史版本的数据？Delta 帮你记着呢。你可以按版本号查，也可以按时间戳查。

# 按版本号读取（版本 0 是初始状态）
df_v0 = spark.read.format("delta").option("versionAsOf", "0").load("/tmp/delta/user_info")
df_v0.show()

# 按时间戳读取
df_time = spark.read.format("delta").option("timestampAsOf", "2024-01-15 10:00:00").load("/tmp/delta/user_info")
df_time.show()

你想想看，这个能力多实用。比如业务方说“昨天的数据不对，我想看看前天跑完的结果”，你直接一个时间旅行查回去，不用再重跑任务。我经常用这个功能做数据对账，省了不少事。

4.3.3 流式读取

Delta 表天然支持流式读取。你可以把它当成一个消息队列来用，新写入的数据能实时读到。

# 流式读取 Delta 表
stream_df = spark.readStream.format("delta").load("/tmp/delta/user_info")

query = stream_df.writeStream \
    .format("console") \
    .outputMode("append") \
    .trigger(processingTime="10 seconds") \
    .start()

query.awaitTermination()

这个流式读取，底层靠的是 Delta 的 Change Data Feed 功能。它只读新增的文件，不会全表扫描。所以性能很好，延迟能做到秒级。

4.4 本章知识体系

我把这章的核心逻辑画了张图，你看一眼就明白了。

4.5 常用操作速查表

最后，我给你整理了一张速查表。平时写代码忘了参数，回来翻翻就行。

操作	API / SQL 示例	说明
创建路径表	`df.write.format("delta").save(path)`	数据存储在指定路径
创建托管表	`df.write.format("delta").saveAsTable(name)`	元数据在 Hive Metastore
SQL 建表	`CREATE TABLE ... USING DELTA`	标准 SQL 语法
追加写入	`.mode("append")`	默认模式，不删旧数据
覆盖写入	`.mode("overwrite")`	替换整个表数据
Upsert	`MERGE INTO ... WHEN MATCHED ...`	按条件更新或插入
批式读取	`spark.read.format("delta").load(path)`	读取最新快照
时间旅行	`.option("versionAsOf", N)`	读取历史版本
流式读取	`spark.readStream.format("delta")`	实时读取新增数据

小技巧：如果你不确定表当前有多少个版本，可以查 DESCRIBE HISTORY table_name。这个命令会列出所有操作记录，包括版本号、操作类型、操作时间。我排查问题时，第一步就是看这个。

好了，Delta 表的基础操作就这些。创建、写入、读取，环环相扣。你把这些练熟了，后面的实时数据湖场景才能玩得转。记住，多练多想，遇到问题先查 _delta_log，那里藏着所有秘密。