📚 课程目录第3章 Delta Lake 基础操作

3. Delta Lake 基础操作：建表、写入、读取与更新删除

好，咱们进入实操环节。Delta Lake 的基础操作，说白了就是围绕「表」做增删改查。但跟普通文件或 Hive 表不一样，Delta Lake 多了一层事务日志（Transaction Log），所以每个操作都带上了 ACID 的「保险」。我刚开始接触时也觉得不就是读写吗？后来发现，理解这些基础操作背后的 Delta 机制，才是用好实时特征的关键。

核心要点：Delta Lake 的所有操作都基于 DeltaTable API 或 Spark SQL。操作结果会先写入事务日志，再更新数据文件。这保证了「读到你该读的，写不丢数据」。

3.1 创建表：两种姿势，一个核心

建表在 Delta Lake 里其实有两种常见方式。我个人习惯在开发阶段用 Spark SQL，因为直观；但在自动化流水线里，我更喜欢用 DataFrame 的 saveAsTable，少写很多 DDL。

方式一：Spark SQL 建表

-- 创建托管表（Managed Table）
CREATE TABLE IF NOT EXISTS user_features (
    user_id STRING,
    feature_name STRING,
    feature_value DOUBLE,
    event_ts TIMESTAMP,
    dt STRING
)
USING DELTA
PARTITIONED BY (dt)
LOCATION '/data/delta/user_features'
TBLPROPERTIES (
    'delta.autoOptimize.optimizeWrite' = 'true',
    'delta.enableChangeDataFeed' = 'true'
);

方式二：DataFrame API 建表

from pyspark.sql import SparkSession

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

# 定义 Schema
schema = "user_id STRING, feature_name STRING, feature_value DOUBLE, event_ts TIMESTAMP, dt STRING"

# 创建空 DataFrame 并保存为 Delta 表
df = spark.createDataFrame([], schema)
df.write.format("delta") \
    .mode("overwrite") \
    .partitionBy("dt") \
    .option("path", "/data/delta/user_features") \
    .saveAsTable("user_features")

我的经验：建表时记得加上 delta.enableChangeDataFeed = 'true'。我在做实时特征回填时，这个属性帮了大忙——它能让你轻松捕获增量变更，不用自己写复杂的增量逻辑。

3.2 写入数据：Append 与 Overwrite 的抉择

写入操作是实时特征落地的日常。你想想看，特征数据源源不断进来，是追加还是覆盖？这里面的门道不少。

Append（追加写入）

适用于增量特征数据，比如每小时的用户行为统计。

# 假设 new_features 是实时计算出来的特征
new_features.write.format("delta") \
    .mode("append") \
    .partitionBy("dt") \
    .save("/data/delta/user_features")

Overwrite（覆盖写入）

适用于全量特征刷新或重跑。但要注意，Overwrite 默认是覆盖整个表，如果你只想覆盖某个分区，得用 replaceWhere。

# 只覆盖 '2024-01-15' 这一天的数据
df.write.format("delta") \
    .mode("overwrite") \
    .option("replaceWhere", "dt = '2024-01-15'") \
    .save("/data/delta/user_features")

避坑指南：我曾经在生产环境里犯过一个错——直接用 mode("overwrite") 写分区表，结果把整个表的历史数据全清空了。幸好 Delta Lake 有事务日志，我用 DESCRIBE HISTORY 找到上一个版本，用 Time Travel 恢复了数据。所以，覆盖分区时务必用 replaceWhere。

3.3 读取数据：Batch Read 与 Time Travel

读取 Delta 表，最常用的就是批读取。但 Delta Lake 的读取有个杀手锏——Snapshot 读取。什么意思？就是你读到的数据是某个时间点的「快照」，不会因为同时有写入而读到一半的数据。

# 标准批读取
df = spark.read.format("delta").load("/data/delta/user_features")
df.show(5)

# 按分区过滤读取（推荐，性能好）
df = spark.read.format("delta").load("/data/delta/user_features") \
    .filter("dt = '2024-01-15'")

# Time Travel：读取历史版本
df_v1 = spark.read.format("delta") \
    .option("versionAsOf", 1) \
    .load("/data/delta/user_features")

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

为什么 Time Travel 重要？在实时特征场景里，模型训练经常需要回溯历史特征。比如你今天发现模型效果不好，想用三天前的特征数据重新训练。没有 Time Travel，你得从原始日志重新跑一遍特征工程。有了它，一行代码搞定。

3.4 更新与删除：Merge 是主角

Delta Lake 的更新和删除，核心就是 MERGE 操作。它也叫「Upsert」（Update + Insert）。实时特征场景里，你经常需要：如果特征已存在则更新，不存在则插入。这就是 MERGE 的典型用法。

MERGE 语法示例

MERGE INTO user_features AS target
USING (
    SELECT 'user_001' AS user_id, 'click_rate' AS feature_name, 0.85 AS feature_value, 
           current_timestamp() AS event_ts, '2024-01-15' AS dt
) AS source
ON target.user_id = source.user_id 
   AND target.feature_name = source.feature_name 
   AND target.dt = source.dt
WHEN MATCHED THEN
    UPDATE SET 
        feature_value = source.feature_value,
        event_ts = source.event_ts
WHEN NOT MATCHED THEN
    INSERT (user_id, feature_name, feature_value, event_ts, dt)
    VALUES (source.user_id, source.feature_name, source.feature_value, source.event_ts, source.dt)

DELETE 操作

删除用得相对少，但清理过期特征时很实用。

# 删除 30 天前的特征数据
DELETE FROM user_features WHERE dt < '2024-01-01'

# 或者用 DataFrame API
from delta.tables import DeltaTable

delta_table = DeltaTable.forPath(spark, "/data/delta/user_features")
delta_table.delete("dt < '2024-01-01'")

性能建议：MERGE 操作在数据量大时可能比较慢。我一般会先确保 ON 条件里的字段有合适的索引或分区裁剪。另外，delta.autoOptimize.optimizeWrite = true 能帮你在写入时做小文件合并，对 MERGE 性能有好处。

3.5 操作对比一览表

操作	适用场景	性能特点	注意事项
CREATE TABLE	初始化特征表	一次性的 DDL 操作	提前规划好分区键和 TBLPROPERTIES
APPEND	增量特征写入	快，只追加新文件	注意小文件问题，定期 OPTIMIZE
OVERWRITE	全量刷新或重跑	中等，会重写数据文件	用 replaceWhere 避免误删全表
BATCH READ	特征查询、训练数据导出	快，支持 Snapshot 和 Time Travel	尽量加分区过滤，减少扫描量
MERGE	Upsert 场景（存在则更新，否则插入）	较慢，需要比对和重写	ON 条件要精确，避免全表扫描
DELETE	清理过期或错误特征	中等，标记删除而非物理删除	删除后运行 VACUUM 才释放存储

嗯，到这里 Delta Lake 的基础操作就讲完了。你会发现，这些操作跟传统数据库很像，但底层实现完全不同。Delta Lake 用事务日志把文件级别的操作「包装」成了表级别的 ACID 语义。你写代码时不用操心文件合并、版本冲突这些破事——Delta Lake 帮你搞定了。

我个人觉得，掌握这些基础操作，你就已经能应付 80% 的实时特征落地场景了。剩下的 20%，比如流写入、Schema Evolution、Z-Order 优化，咱们后面章节再聊。