📚 课程目录第4章基础读写操作

4. 基础读写操作：创建Delta表、写入数据（Append/Overwrite）、读取数据、更新与删除

好，咱们进入实战环节。前面聊了那么多 Delta Lake 的原理和架构，说白了，最终都得落到代码上。这一章我带你亲手操作一下——从零创建一张 Delta 表，然后往里写数据、读数据，再试试更新和删除。

这些操作看着简单，但里面有不少门道。我当年第一次用 Delta Lake 时，就因为在 Overwrite 模式下没指定分区，差点把生产数据给清了。嗯，咱们一步步来，把每个操作都吃透。

4.1 创建 Delta 表

创建 Delta 表的方式有好几种。我个人习惯用 Spark SQL 的 CREATE TABLE 语法，因为最直观。当然，你也可以用 DataFrame 的 save() 方法，效果一样。

4.1.1 使用 Spark SQL 创建

-- 创建一张用户行为日志表
CREATE TABLE IF NOT EXISTS user_events (
    event_id STRING,
    user_id STRING,
    event_type STRING,
    event_time TIMESTAMP,
    page_url STRING
)
USING DELTA
PARTITIONED BY (event_date STRING)
LOCATION '/data/delta/user_events'

这里要注意几点：

USING DELTA 是必须的，告诉 Spark 用 Delta Lake 引擎
PARTITIONED BY 按日期分区，这是数仓建模的常规操作
LOCATION 指定表数据存储路径，如果不指定，会写到 Spark 默认的仓库目录

我的习惯：创建表时一定指定 LOCATION。这样表和数据是解耦的，哪天要迁移或重建元数据，直接指向路径就行，不用重新导入数据。

4.1.2 使用 DataFrame API 创建

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("delta_basics") \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
    .getOrCreate()

# 从 DataFrame 创建 Delta 表
df = spark.range(0, 1000) \
    .withColumn("data", (spark.range(0, 1000).col("id") * 2).cast("int"))

df.write.format("delta") \
    .mode("overwrite") \
    .save("/data/delta/simple_table")

你看，用 DataFrame 的方式更灵活。你可以先对数据进行各种转换，最后一把写入。我在项目中经常这么干——先 ETL 清洗，再写到 Delta 表。

4.2 写入数据：Append 与 Overwrite

写入模式是 Delta Lake 的核心能力之一。说白了就两种：Append（追加）和 Overwrite（覆盖）。但用不好，坑很大。

模式	行为	适用场景	风险
Append	在现有数据后追加新数据	日志接入、实时流写入	低，但可能产生小文件
Overwrite	删除旧数据，写入新数据	全量刷新、数据重跑	高，误操作会丢数据

4.2.1 Append 模式

# 模拟一批新的事件数据
new_events = spark.createDataFrame([
    ("evt1001", "user001", "click", "2024-01-15 10:00:00", "/home", "2024-01-15"),
    ("evt1002", "user002", "view", "2024-01-15 10:01:00", "/product", "2024-01-15"),
], schema="event_id STRING, user_id STRING, event_type STRING, event_time STRING, page_url STRING, event_date STRING")

new_events.write.format("delta") \
    .mode("append") \
    .save("/data/delta/user_events")

Append 模式很简单，就是往表里加数据。我经常用它来做增量接入——每天跑一次任务，把当天的日志追加进去。

我曾经踩过的坑：Append 模式如果频繁写入小批次数据，会产生大量小文件。Delta Lake 虽然会自动做小文件合并（OPTIMIZE），但最好还是控制写入频率，比如每 5-10 分钟一批，而不是每秒一批。

4.2.2 Overwrite 模式

# 全量覆盖某一天的数据
df_day = spark.read.format("delta") \
    .load("/data/delta/user_events") \
    .filter("event_date = '2024-01-15'")

# 对数据进行重新处理
df_corrected = df_day.withColumn("event_type", 
    when(col("event_type") == "click", "click_v2").otherwise(col("event_type")))

# 覆盖写入
df_corrected.write.format("delta") \
    .mode("overwrite") \
    .option("replaceWhere", "event_date = '2024-01-15'") \
    .save("/data/delta/user_events")

这里有个关键点：replaceWhere 选项。它允许你只覆盖特定分区的数据，而不是整张表。你想想看，如果没有这个选项，一个 Overwrite 下去，整张表就只剩你当前写入的数据了——其他分区的数据全没了。

    最佳实践：使用 Overwrite 模式时，永远带上 replaceWhere 条件。除非你确定要全表覆盖，否则别裸写 Overwrite。

4.3 读取数据

读取 Delta 表的方式也很灵活。你可以读全表，也可以按分区、按时间戳、按版本读。这就是 Delta Lake 的 Time Travel 能力。

4.3.1 基础读取

# 读取全表
df = spark.read.format("delta").load("/data/delta/user_events")

# 按分区过滤（分区裁剪）
df_filtered = spark.read.format("delta") \
    .load("/data/delta/user_events") \
    .filter("event_date = '2024-01-15'")

# 只读特定列（列裁剪）
df_columns = spark.read.format("delta") \
    .load("/data/delta/user_events") \
    .select("event_id", "user_id", "event_type")

分区裁剪和列裁剪是性能优化的两个基本手段。我在项目中见过有人直接读全表，然后 filter——数据量小还好，要是几百 TB 的表，那得等到天荒地老。

4.3.2 Time Travel 读取

# 按时间戳读取历史版本
df_as_of_time = spark.read.format("delta") \
    .option("timestampAsOf", "2024-01-14 23:59:59") \
    .load("/data/delta/user_events")

# 按版本号读取
df_version_5 = spark.read.format("delta") \
    .option("versionAsOf", 5) \
    .load("/data/delta/user_events")

Time Travel 这个功能，说白了就是后悔药。有一次我在生产环境跑错了一个任务，把某天的数据写坏了。要是普通表，只能从备份恢复，至少折腾半小时。但用 Delta Lake，我直接 timestampAsOf 读到前一天的状态，然后重新跑任务——前后不到 5 分钟。

4.4 更新与删除

Delta Lake 支持标准的 UPDATE、DELETE、MERGE 操作。注意，这些操作不是原地修改数据文件，而是通过事务日志记录变更，生成新版本的文件。所以，它们都是 ACID 的。

4.4.1 UPDATE 操作

# 使用 Spark SQL 更新
spark.sql("""
    UPDATE user_events
    SET event_type = 'converted_click'
    WHERE event_type = 'click' AND event_date = '2024-01-15'
""")

# 使用 DataFrame API 更新
from delta.tables import DeltaTable

delta_table = DeltaTable.forPath(spark, "/data/delta/user_events")

delta_table.update(
    condition = "event_type = 'click' AND event_date = '2024-01-15'",
    set = { "event_type": "'converted_click'" }
)

我个人更推荐用 DataFrame API 的方式，因为可以跟 Spark 的 DataFrame 操作无缝衔接。比如你可以先读一个 DataFrame，做复杂逻辑处理，然后用 merge 操作写回去。

4.4.2 DELETE 操作

# 删除特定条件的数据
delta_table.delete(
    condition = "event_date < '2024-01-01'"
)

# 也可以按版本删除（逻辑删除）
# 实际上是通过 Time Travel 恢复到旧版本
spark.read.format("delta") \
    .option("versionAsOf", 3) \
    .load("/data/delta/user_events") \
    .write.format("delta") \
    .mode("overwrite") \
    .save("/data/delta/user_events")

注意：DELETE 操作不会立即释放磁盘空间。被删除的数据文件只是被标记为"不再使用"，但物理文件还在。要真正回收空间，需要执行 VACUUM 命令。不过 VACUUM 有默认保留期（7天），确保你有足够的时间做 Time Travel。

4.4.3 MERGE 操作（UPSERT）

MERGE 是 Delta Lake 最强大的操作之一。它可以根据条件，在一张表上同时执行 INSERT、UPDATE、DELETE。说白了就是"有则更新，无则插入"。

# 准备一批新的数据（可能包含已存在的记录）
new_data = spark.createDataFrame([
    ("evt1001", "user001", "click_v2", "2024-01-15 10:00:00", "/home", "2024-01-15"),
    ("evt1003", "user003", "purchase", "2024-01-15 11:00:00", "/checkout", "2024-01-15"),
], schema="event_id STRING, user_id STRING, event_type STRING, event_time STRING, page_url STRING, event_date STRING")

# 执行 MERGE
delta_table.alias("target") \
    .merge(
        source=new_data.alias("source"),
        condition="target.event_id = source.event_id"
    ) \
    .whenMatchedUpdate(set={
        "event_type": "source.event_type",
        "event_time": "source.event_time",
        "page_url": "source.page_url"
    }) \
    .whenNotMatchedInsert(values={
        "event_id": "source.event_id",
        "user_id": "source.user_id",
        "event_type": "source.event_type",
        "event_time": "source.event_time",
        "page_url": "source.page_url",
        "event_date": "source.event_date"
    }) \
    .execute()

这个操作在 CDC（变更数据捕获）场景下特别有用。我曾经帮一个金融客户做实时数仓，他们每天从业务库同步增量数据，就用 MERGE 来保证目标表和源表一致。效率比先删后插高了一个数量级。

4.5 本章知识体系

下面这张图总结了 Delta Lake 基础读写操作的核心流程和关系：

从这张图可以看得很清楚：创建表是起点，写入数据有 Append 和 Overwrite 两种模式，读取数据支持多种方式，而更新、删除、合并操作都通过事务日志来保证一致性。

4.6 避坑指南

最后，我把自己这些年踩过的坑总结一下，你遇到了可以少走弯路：

Overwrite 不带 replaceWhere：我曾经有一次在测试环境跑 Overwrite，忘了加分区条件，结果把整张表覆盖了。还好是测试环境，不然就等着被老板约谈了。
频繁 Append 产生小文件：Delta Lake 虽然会自动做小文件合并，但如果你每秒 Append 一次，合并速度跟不上写入速度。建议控制写入频率，或者用 coalesce() 合并分区。
DELETE 后空间不释放：记得定期执行 VACUUM，但保留期别设太短。我一般保留 7 天，这样万一需要 Time Travel 恢复数据，还有机会。
MERGE 条件写错：MERGE 的 condition 如果写得太宽泛，可能会匹配到不该匹配的行。我建议先跑一个 SELECT 验证条件，确认没问题再执行 MERGE。

好了，这一章的内容就到这里。Delta Lake 的基础读写操作其实不难，但细节决定成败。你把这些操作练熟了，后面讲 Schema Evolution、Optimize、Z-Order 这些高级功能时，才能游刃有余。