3. Delta Lake 基础操作：表的创建、读写、更新、删除，ACID 事务初体验
好，咱们正式开始动手了。前面两章聊了 Delta Lake 的定位和架构，说白了都是「纸上谈兵」。这一章，咱们直接上代码，把 Delta 表的基础操作跑一遍。
我个人习惯，学一个新东西，先把它当成一个「超级数据库表」来用。创建、写入、读取、更新、删除，这五板斧抡熟了，后面再聊什么时间旅行、Schema 演化，心里才有底。
嗯，这里要提前说一句：Delta Lake 本质上是一个存储层，它依赖 Spark 来执行计算。所以下面的代码，你需要在 Spark Shell 或者 Databricks Notebook 里跑。我假设你已经搭好了 Spark 环境，并且引入了 Delta Lake 的依赖包。
3.1 创建 Delta 表：两种方式，一个核心
创建 Delta 表，说白了就两种路子：
- 从 DataFrame 直接保存：你手头有数据，想存成 Delta 格式。
- 先建目录，再写数据：你先定义好表结构，后面再往里灌数据。
我在项目中遇到过不少新手，上来就用第二种，结果发现 Schema 对不上，折腾半天。我建议你，初期就用第一种，让 Spark 帮你推断 Schema，省心。
3.1.1 方式一：从 DataFrame 创建
// 1. 造一点示例数据
val data = Seq(
("user_001", "click", 1625000000L, 0.5),
("user_002", "view", 1625000060L, 0.8),
("user_003", "purchase", 1625000120L, 20.0)
)
// 2. 定义 Schema（我习惯显式指定，避免类型推断翻车）
import org.apache.spark.sql.types._
val schema = StructType(Array(
StructField("user_id", StringType, true),
StructField("event_type", StringType, true),
StructField("ts", LongType, true),
StructField("value", DoubleType, true)
))
// 3. 创建 DataFrame
val df = spark.createDataFrame(data).toDF(schema.fieldNames: _*)
// 4. 保存为 Delta 表
df.write
.format("delta")
.mode("overwrite") // 第一次用 overwrite，后面增量用 append
.save("/tmp/delta_events")
你看，就这么简单。执行完后，去 /tmp/delta_events 目录下看看，你会发现多了一个 _delta_log 文件夹。这就是 Delta Lake 的「心脏」——事务日志。
3.1.2 方式二：先定义表，再插入数据
这种方式更接近传统数据库的 CREATE TABLE。适合你已经有明确的 Schema 设计，或者需要跟 Hive Metastore 集成的时候。
// 1. 创建空表
spark.sql("""
CREATE TABLE IF NOT EXISTS delta_events (
user_id STRING,
event_type STRING,
ts LONG,
value DOUBLE
) USING DELTA
LOCATION '/tmp/delta_events_sql'
""")
// 2. 插入数据
spark.sql("""
INSERT INTO delta_events VALUES
('user_001', 'click', 1625000000, 0.5),
('user_002', 'view', 1625000060, 0.8)
""")
3.2 读取 Delta 表：快照读，还是流式读？
读取 Delta 表，最常用的就是 spark.read.format("delta").load(path)。但这里有个关键点：你读到的，是表在当前时间点的「快照」。
// 读取全量数据
val df = spark.read.format("delta").load("/tmp/delta_events")
df.show()
// 你也可以用 SQL 查（如果注册了临时视图）
df.createOrReplaceTempView("events")
spark.sql("SELECT user_id, event_type, value FROM events WHERE value > 1.0").show()
为什么会强调「快照」？因为 Delta Lake 的 ACID 事务保证了，你每次读到的数据都是一致的。哪怕有人在同时写，你也不会读到一半的数据。这一点，我在做实时特征 pipeline 的时候感受特别深——以前用 Parquet 直接覆盖，查询经常报错，换成 Delta 之后，再也没出现过「文件找不到」的诡异问题。
3.3 更新和删除：终于可以改数据了
用过 Hive 或者 Spark Parquet 的同学都知道，想改一条数据，得把整个分区重写一遍，又慢又容易出错。Delta Lake 支持标准的 UPDATE、DELETE、MERGE 操作，这才是它真正值钱的地方。
3.3.1 UPDATE 操作
// 把 user_001 的 event_type 改成 'special_click'
import io.delta.tables._
val deltaTable = DeltaTable.forPath(spark, "/tmp/delta_events")
deltaTable.update(
condition = expr("user_id = 'user_001'"),
set = Map("event_type" -> expr("'special_click'"))
)
// 验证结果
spark.read.format("delta").load("/tmp/delta_events").show()
3.3.2 DELETE 操作
// 删除 value 小于 1.0 的事件
deltaTable.delete(condition = expr("value < 1.0"))
3.3.3 MERGE 操作（UPSERT）
这是实时特征工程的王牌操作。你想啊，特征数据源源不断地进来，有些是新用户，有些是老用户需要更新特征值。用 MERGE 一句搞定：
// 假设这是新来的增量数据
val updatesDF = Seq(
("user_001", "click", 1625100000L, 0.9), // 更新已有用户
("user_004", "view", 1625100060L, 0.3) // 新用户
).toDF("user_id", "event_type", "ts", "value")
deltaTable.as("target")
.merge(
updatesDF.as("source"),
"target.user_id = source.user_id"
)
.whenMatched()
.updateExpr(Map(
"event_type" -> "source.event_type",
"ts" -> "source.ts",
"value" -> "source.value"
))
.whenNotMatched()
.insertAll()
.execute()
target.user_id = source.user_id AND target.ts = source.ts，结果因为时间戳对不上，所有数据都变成了 INSERT，导致大量重复。后来我学乖了，匹配条件尽量用业务主键，比如 user_id + event_type。
3.4 ACID 事务初体验：看看事务日志里有什么
说了半天 ACID，到底长什么样？咱们直接去 _delta_log 目录里看一眼。
// 查看事务日志文件
import org.apache.hadoop.fs._
val fs = FileSystem.get(spark.sparkContext.hadoopConfiguration)
val logPath = new Path("/tmp/delta_events/_delta_log")
val files = fs.listStatus(logPath)
files.foreach(f => println(f.getPath.getName))
你会看到一堆 00000000000000000000.json、00000000000000000001.json 这样的文件。每个文件对应一次事务提交。打开第一个看看：
// 读取第一个事务日志
val logDF = spark.read.json("/tmp/delta_events/_delta_log/00000000000000000000.json")
logDF.show(truncate = false)
你会看到类似这样的内容：
| add | remove | commitInfo | protocol | metaData |
|---|---|---|---|---|
| {path, size, partitionValues, ...} | null | {operation, timestamp, ...} | {minReaderVersion, minWriterVersion} | {id, name, schema, ...} |
这就是 Delta Lake 的「账本」。每次写入、更新、删除，都会生成一个新的事务日志，记录哪些文件被添加、哪些被删除。读取的时候，Spark 只读最新的「快照」——也就是所有 add 操作对应的文件，跳过那些被 remove 的。
_delta_log 实现的。它把「数据文件」和「元数据」分开了。元数据（事务日志）是原子的、一致的、隔离的、持久的。数据文件本身是不可变的，每次修改都是生成新文件，旧文件在日志里标记为删除。
嗯，说到这，我想起之前有个同事问我：「Delta Lake 的 UPDATE 是不是真的改文件？」我让他去看了事务日志。他看完恍然大悟——原来 UPDATE 不是改文件，而是写新文件、删旧文件。这个理解很重要，因为它解释了为什么 Delta Lake 能支持时间旅行（读历史版本），也解释了为什么频繁 UPDATE 会导致小文件问题。
3.5 本章小结：你其实已经掌握了核心
这一章的内容，说白了就是四个字：增删改查。但 Delta Lake 的「改」和「删」，跟传统 Spark 不一样，它是带事务的。你每执行一次操作，背后都有一笔「账」被记下来。
我建议你，学完这一章，立刻去跑一遍代码。重点看两个东西：
- 事务日志：每次操作后，去
_delta_log目录下看看多了哪个文件，里面写了什么。 - 数据目录：看看
.snappy.parquet文件的数量是怎么变的。UPDATE 一次，是不是多了新文件？
把这些看明白了，Delta Lake 在你眼里就不再是一个黑盒了。下一章，咱们会在这个基础上，聊一个更酷的特性——时间旅行。但那是后话了，先把基础打牢。