环境搭建与入门：Spark 与 Delta Lake 集成、本地开发环境配置、第一个 Delta 表

好，咱们正式开始动手。上一章聊了 Delta Lake 到底解决了什么问题，这一章就直接把环境搭起来，跑通第一个 Delta 表。我个人习惯是先把开发环境配到「能跑」的状态，再慢慢调优。别一上来就搞集群，本地先跑通，心里才有底。

2.1 整体架构：Spark + Delta Lake 怎么配合

先看一张图，搞清楚我们今天要搭的东西长什么样。

说白了，Delta Lake 不是一个独立服务，它是 Spark 的一个插件。你启动 SparkSession 的时候带上 Delta 的扩展包，就能用上事务日志、ACID、时间旅行这些能力。我在项目中遇到过有人把 Delta 当成独立数据库来部署，折腾半天发现跑不起来——其实它就是个库，寄生在 Spark 里。

2.2 本地开发环境配置

我建议你直接用 IntelliJ IDEA + Maven 或者 VS Code + sbt，看个人喜好。下面以 Maven 项目为例，一步步来。

2.2.1 依赖与版本选择

版本匹配是新手最容易踩坑的地方。Delta Lake 和 Spark 的版本有严格的对应关系。我曾经因为用了 Delta 2.0 配 Spark 3.1，结果一堆 API 找不到，查了两天文档才发现是版本不兼容。

这里给出一组经过验证的版本组合：

2.2.2 Maven 依赖配置

在 pom.xml 里加入以下核心依赖：

<properties>
    <scala.version>2.12.17</scala.version>
    <spark.version>3.3.2</spark.version>
    <delta.version>2.3.0</delta.version>
</properties>

<dependencies>
    <!-- Spark Core -->
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_2.12</artifactId>
        <version>${spark.version}</version>
    </dependency>
    
    <!-- Spark SQL -->
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-sql_2.12</artifactId>
        <version>${spark.version}</version>
    </dependency>
    
    <!-- Delta Lake 核心包 -->
    <dependency>
        <groupId>io.delta</groupId>
        <artifactId>delta-core_2.12</artifactId>
        <version>${delta.version}</version>
    </dependency>
</dependencies>

2.2.3 初始化 SparkSession 并启用 Delta

这是最关键的一步。Delta Lake 通过 Spark 的扩展机制加载，你需要在创建 SparkSession 时显式指定。

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder()
    .appName("DeltaLakeDemo")
    .master("local[*]")  // 本地模式，使用所有可用核心
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
    .getOrCreate()

println("SparkSession 创建成功，Delta Lake 已启用！")

嗯，这里要注意两个配置项：

• spark.sql.extensions：注册 Delta 的 SQL 扩展，让 Spark 能识别 Delta 表的语法。
• spark.sql.catalog.spark_catalog：把默认的 catalog 替换成 DeltaCatalog，这样 Spark 才知道 Delta 表怎么读写。

我刚开始学的时候，只配了第一个，忘了配第二个，结果能建表但查不了数据——你想想看，这多憋屈。

2.3 第一个 Delta 表

环境搭好了，咱们来写第一个 Delta 表。我会用 Scala 和 SQL 两种方式演示，你选一种顺手的学习就行。

2.3.1 用 Scala API 创建 Delta 表

// 1. 创建示例数据
import spark.implicits._

val data = Seq(
    ("user001", "Alice", 28, "2024-01-15"),
    ("user002", "Bob", 35, "2024-01-16"),
    ("user003", "Charlie", 22, "2024-01-17")
).toDF("user_id", "name", "age", "reg_date")

// 2. 保存为 Delta 表
data.write
    .format("delta")
    .mode("overwrite")
    .save("/tmp/delta_user_demo")

println("Delta 表创建成功！路径：/tmp/delta_user_demo")

就这么简单。运行之后，去 /tmp/delta_user_demo 目录看看，你会发现除了 Parquet 数据文件，还多了一个 _delta_log 文件夹。这就是 Delta Lake 的事务日志，里面记录着每一次操作。

2.3.2 用 SQL 方式创建 Delta 表

如果你更习惯 SQL，Delta Lake 也支持标准 DDL：

// 先注册一个临时视图
data.createOrReplaceTempView("user_temp")

// 用 SQL 创建 Delta 表
spark.sql("""
    CREATE TABLE delta_user_sql
    USING delta
    LOCATION '/tmp/delta_user_sql'
    AS SELECT * FROM user_temp
""")

// 查询验证
spark.sql("SELECT * FROM delta_user_sql").show()

2.3.3 验证 Delta 表的核心特性

表建好了，咱们验证一下 Delta Lake 最核心的两个能力：时间旅行 和 数据版本。

// 1. 查看表的历史版本
spark.sql("DESCRIBE HISTORY delta_user_sql").show(false)

// 输出示例：
// +-------+-------------------+-------+--------+------------------+
// |version|timestamp          |userId |operation|operationParameters|
// +-------+-------------------+-------+--------+------------------+
// |0      |2024-01-20 10:30:00|...    |WRITE   |{mode=Overwrite,...}|
// +-------+-------------------+-------+--------+------------------+

// 2. 更新数据
spark.sql("UPDATE delta_user_sql SET age = 29 WHERE user_id = 'user001'")

// 3. 再次查看历史，版本变成了 1
spark.sql("DESCRIBE HISTORY delta_user_sql").show(false)

// 4. 时间旅行：查询版本 0 的数据（更新前的数据）
spark.sql("SELECT * FROM delta_user_sql VERSION AS OF 0").show()

看到了吗？每次写入、更新、删除操作都会产生一个新版本。你可以随时回到任意历史版本查数据。我在项目中就用这个特性做过数据回滚——有一次凌晨跑批任务把线上表写坏了，直接一个 RESTORE 命令回到前一版本，五分钟解决问题，不用找 DBA 从备份恢复。

2.4 避坑指南

最后分享几个我踩过的坑，你遇到了可以直接翻回来查：

• 路径问题：Delta 表路径不要包含中文或特殊字符，否则在某些文件系统上会报错。
• 并发写入：本地开发时不要同时开多个 SparkSession 写同一个 Delta 表，会触发并发冲突。生产环境有专门的并发控制机制，后面章节会讲。
• 清理临时文件：Delta 表不会自动清理旧版本数据，本地测试时记得定期 VACUUM，否则磁盘会爆。我有个同事就是忘了这茬，跑了一周测试，磁盘直接满了。

好了，环境搭好了，第一个 Delta 表也跑通了。你可以试着往表里多写几批数据，看看版本号怎么递增，再试试时间旅行查不同版本的数据。动手试一遍，比看十遍文档都管用。