3. Delta Lake 基础操作：创建表、写入数据、读取数据、Schema 演化

好，咱们进入实战环节。Delta Lake 的基础操作，说白了就是四个动作：建表、写数据、读数据、改 Schema。我刚开始接触 Delta 时，觉得这不就是 Spark SQL 嘛，后来踩了几个坑才发现，细节里全是魔鬼。

今天咱们把这四个动作拆开揉碎，每个操作我都会带上实际项目里的教训。你跟着走一遍，基本就能应付日常开发了。

3.1 创建 Delta 表：两种方式，一个核心

创建 Delta 表有两种主流方式。我个人习惯用 DataFrame API，因为灵活；但如果你团队里分析师多，SQL 方式更友好。

3.1.1 方式一：DataFrame API 建表

from pyspark.sql import SparkSession

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

# 定义 Schema
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, TimestampType

schema = StructType([
    StructField("user_id", StringType(), False),
    StructField("user_name", StringType(), True),
    StructField("age", IntegerType(), True),
    StructField("event_time", TimestampType(), True)
])

# 创建空 DataFrame 并写入 Delta 格式
data = []
df = spark.createDataFrame(data, schema)
df.write.format("delta").save("/data/delta/users")
我的习惯：建表时一定指定 StructField 的 nullable 参数。主键字段设为 False，其他字段尽量 True。为什么？因为数据源经常有脏数据，非空约束能帮你提前发现问题。

3.1.2 方式二：SQL 方式建表

spark.sql("""
    CREATE TABLE IF NOT EXISTS delta.`/data/delta/users` (
        user_id STRING NOT NULL,
        user_name STRING,
        age INT,
        event_time TIMESTAMP
    ) USING DELTA
""")

嗯，这里要注意：SQL 方式建表时，USING DELTA 不能写成 USING DELTA LAKE。我见过有同事写错了，结果 Spark 默认用了 Parquet 格式，事务日志全没了。

3.2 写入数据：Append、Overwrite、Upsert

写入数据是 Delta 的核心能力。三种模式各有适用场景，我按项目经验给你排个优先级。

写入模式 语法 适用场景 注意事项
Append df.write.mode("append") 日志、事件流、CDC 增量 不会覆盖已有数据
Overwrite df.write.mode("overwrite") 全量刷新、重跑任务 会删除旧数据，慎用
Upsert DeltaTable.merge() 维表更新、CDC 合并 需要指定匹配条件

3.2.1 Append 模式：最常用

# 模拟一批新用户数据
new_users = [
    ("U001", "张三", 28, "2024-01-15 10:00:00"),
    ("U002", "李四", 35, "2024-01-15 10:05:00")
]
columns = ["user_id", "user_name", "age", "event_time"]
df_new = spark.createDataFrame(new_users, columns)

df_new.write.format("delta").mode("append").save("/data/delta/users")
我曾经踩过的坑：Append 模式下，如果 DataFrame 的 Schema 和目标表不一致，Delta 默认会报错。比如你多了一个字段，或者字段类型不匹配，写入会直接失败。解决办法是开启 Schema 自动演化，后面会讲。

3.2.2 Overwrite 模式：全量替换

# 全量覆盖用户表
df_full = spark.read.parquet("/data/raw/users_full_20240115")
df_full.write.format("delta").mode("overwrite").save("/data/delta/users")

Overwrite 模式会替换整个表的数据，但事务日志会保留。这意味着你可以用 DESCRIBE HISTORY 看到每次覆盖的记录。我有个项目就是靠这个功能回滚了误操作。

3.2.3 Upsert 模式：Merge 操作

from delta.tables import DeltaTable

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

# 模拟 CDC 数据
cdc_data = [
    ("U001", "张三丰", 29, "2024-01-16 08:00:00"),  # 更新
    ("U003", "王五", 40, "2024-01-16 08:00:00")     # 新增
]
df_cdc = spark.createDataFrame(cdc_data, columns)

delta_table.alias("target") \
    .merge(df_cdc.alias("source"), "target.user_id = source.user_id") \
    .whenMatchedUpdateAll() \
    .whenNotMatchedInsertAll() \
    .execute()

Merge 操作是 CDC 入湖的核心。你想想看，如果没有这个能力，每次更新都要全表覆盖，那效率得多低。

3.3 读取数据：时间旅行与增量读

Delta 的读取能力比普通 Parquet 强太多了。我最喜欢的是 时间旅行 和 增量读，这两个功能在数据回溯和流处理中特别有用。

3.3.1 基础读取

# 读取最新版本
df = spark.read.format("delta").load("/data/delta/users")
df.show()

3.3.2 时间旅行：回到过去

# 按版本号读取
df_v1 = spark.read.format("delta") \
    .option("versionAsOf", 1) \
    .load("/data/delta/users")

# 按时间戳读取
df_time = spark.read.format("delta") \
    .option("timestampAsOf", "2024-01-15 12:00:00") \
    .load("/data/delta/users")
为什么需要时间旅行？ 我遇到过好几次：业务方说「昨天的报表数据不对，能不能看看三天前的数据长什么样？」。有了时间旅行，直接按版本号读就行，不用再跑一次历史任务。

3.3.3 增量读取：只读新数据

# 从指定版本开始增量读
df_incremental = spark.read.format("delta") \
    .option("readChangeFeed", "true") \
    .option("startingVersion", 5) \
    .load("/data/delta/users")

df_incremental.select("user_id", "user_name", "_change_type", "_commit_version").show()

增量读配合 CDC 使用，效果绝佳。你只需要记录上次处理的版本号，下次从那个版本开始读，就能拿到所有变更数据。

3.4 Schema 演化：让表结构活起来

Schema 演化是 Delta 最实用的特性之一。说白了，就是允许表结构随着数据源的变化而自动调整。我刚开始做数据湖时，最头疼的就是上游加字段，下游就要改代码。有了 Schema 演化，这个问题基本解决了。

3.4.1 自动演化：加字段

# 新数据多了一个 phone 字段
new_data_with_phone = [
    ("U004", "赵六", 32, "2024-01-17 09:00:00", "13800138000")
]
columns_with_phone = ["user_id", "user_name", "age", "event_time", "phone"]
df_with_phone = spark.createDataFrame(new_data_with_phone, columns_with_phone)

# 开启 Schema 演化
df_with_phone.write.format("delta") \
    .option("mergeSchema", "true") \
    .mode("append") \
    .save("/data/delta/users")

执行完后，表里会自动多出 phone 字段，旧数据的 phone 值为 null。这就是自动演化的威力。

3.4.2 手动演化：改字段类型

from delta.tables import DeltaTable

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

# 将 age 字段从 IntegerType 改为 LongType
delta_table.alterTableAddColumns(
    "age BIGINT"
)
注意：Delta 不支持所有类型的自动转换。比如 StringType 转 IntegerType 会失败，因为数据可能不兼容。我建议手动演化时，先检查数据质量，再执行变更。

3.4.3 Schema 演化的最佳实践

  • 生产环境默认关闭自动演化：用 mergeSchema=false，避免意外加字段导致下游任务报错。
  • 加字段用 Append 模式：Overwrite 模式会覆盖整个表，Schema 演化效果不如 Append 直观。
  • 删字段要谨慎：Delta 不支持直接删除字段，只能通过重写表来实现。我建议用 SELECT 排除不需要的字段，而不是物理删除。

3.5 本章知识体系图

下面这张图把 Delta Lake 基础操作的四个核心环节串起来了。你可以看到，建表是起点，写入和读取是双向通道，Schema 演化贯穿始终。

Delta Lake 基础操作知识体系 Delta Lake 基础操作知识体系 Delta 表 创建表 写入数据 读取数据 Schema 演化 核心要点 建表是基础，写入和读取是双向通道，Schema 演化贯穿始终 时间旅行和增量读是 Delta 区别于普通 Parquet 的关键特性

3.6 避坑指南：我踩过的三个坑

  1. Schema 演化导致下游任务崩溃：有一次我开启了自动演化，上游加了个字段，下游的 ETL 任务直接报错，因为字段顺序变了。后来我改成手动演化，加字段前先通知下游团队。
  2. Merge 操作性能问题：当表数据量超过 10 亿行时，Merge 操作会变得很慢。我的解决方案是：先做分区裁剪，只 Merge 受影响的分区。
  3. 时间旅行版本保留时间：Delta 默认保留 30 天的历史版本。如果你需要更长的回溯能力，记得调大 delta.logRetentionDuration 参数。
我的建议：刚开始用 Delta 时，先在小数据量上把 Append、Overwrite、Merge 三种模式都跑一遍，观察事务日志的变化。理解了日志机制，后面遇到问题就能快速定位。

好了，Delta Lake 的基础操作就讲到这里。你把这些代码跑一遍，再结合自己的业务场景想一想，应该就能上手了。记住，建表时想清楚 Schema，写入时选对模式，读取时用好时间旅行，Schema 演化要谨慎。这四个点抓住了，Delta Lake 的基本功就算扎实了。

← 上一章 📖 返回目录 下一章 →