3、环境搭建：Spark + Delta Lake 本地开发环境配置

说实话，很多同学学 Delta Lake 卡住的第一步，不是概念难懂，而是环境没配好。

我见过太多人花了两天装 Spark，最后跑个 read.format("delta") 就报 ClassNotFoundException。嗯，这章我们就来解决这个问题。

我会带你从零搭一套能跑 Delta Lake 的本地环境。你跟着走一遍，后面所有实战都能直接跑。

3.1 环境总览：我们需要什么？

先看整体结构。本地开发环境说白了就三层：

• 底层：Java 8/11 + Scala 2.12/2.13 —— Spark 跑在 JVM 上，这是基础。
• 中间层：Apache Spark 3.x —— 计算引擎，Delta Lake 是它的扩展。
• 上层：Delta Lake 库 —— 提供 ACID、时间旅行、Schema 强制等功能。

我个人的习惯是：用 Maven 或 SBT 管理依赖，而不是手动下载 Jar 包。为什么？因为版本匹配太容易出错了。你想想看，Spark 3.2 配 Delta 1.0 能跑，但 Spark 3.3 配 Delta 2.0 可能就报错。用构建工具，版本号写死，省心。

下面这张图帮你理清各组件的关系：

3.2 安装 Java 和 Scala

先检查你的机器上有没有 Java。打开终端：

java -version
# 应该输出类似：openjdk version "1.8.0_382" 或 "11.0.20"

如果没有，去 Adoptium 下载 JDK 8 或 11。我个人推荐 JDK 11，Spark 3.5 对 JDK 11 支持最好。

Scala 呢？其实你不需要单独装 Scala 编译器。Spark 自带了 Scala 运行时。但如果你要用 SBT 构建项目，那需要装 Scala。不过我们这章用 Maven，所以 Scala 版本由 Spark 决定——Spark 3.5 用的是 Scala 2.12。

3.3 安装 Apache Spark

这一步我踩过坑。以前我直接下载预编译包，结果 Hadoop 版本不对，读本地文件都报错。后来我学乖了：

1. 去 Spark 官网 下载 spark-3.5.0-bin-hadoop3.tgz
2. 解压到 /opt/spark 或 C:\spark
3. 设置环境变量 SPARK_HOME 和 PATH

Linux/Mac 上：

export SPARK_HOME=/opt/spark/spark-3.5.0-bin-hadoop3
export PATH=$SPARK_HOME/bin:$PATH

Windows 上：在系统环境变量里加 SPARK_HOME，然后把 %SPARK_HOME%\bin 加到 Path。

验证安装：

spark-shell --version
# 应该输出 Spark version 3.5.0

3.4 创建 Maven 项目并引入 Delta Lake

我个人习惯用 Maven 管理依赖。创建一个标准 Maven 项目，在 pom.xml 里加这些：

<properties>
    <scala.version>2.12.18</scala.version>
    <spark.version>3.5.0</spark.version>
    <delta.version>3.1.0</delta.version>
</properties>

<dependencies>
    <!-- Spark Core -->
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_2.12</artifactId>
        <version>${spark.version}</version>
        <scope>provided</scope>
    </dependency>

    <!-- Spark SQL -->
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-sql_2.12</artifactId>
        <version>${spark.version}</version>
        <scope>provided</scope>
    </dependency>

    <!-- Delta Lake -->
    <dependency>
        <groupId>io.delta</groupId>
        <artifactId>delta-core_2.12</artifactId>
        <version>${delta.version}</version>
    </dependency>

    <!-- Delta Lake 独立表 API（可选） -->
    <dependency>
        <groupId>io.delta</groupId>
        <artifactId>delta-standalone_2.12</artifactId>
        <version>3.0.0</version>
    </dependency>
</dependencies>

注意 scope=provided 的意思是：编译时需要，但运行时由 Spark 提供。这样打包时不会把 Spark 的 Jar 打进去，避免冲突。

3.5 编写第一个 Delta Lake 程序

环境搭好了，我们来写个最简单的程序验证一下。创建一个 Scala 对象：

import org.apache.spark.sql.SparkSession

object DeltaQuickStart {
  def main(args: Array[String]): Unit = {
    // 创建 SparkSession，启用 Delta Lake 扩展
    val spark = SparkSession.builder()
      .appName("DeltaQuickStart")
      .master("local[*]")
      .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
      .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
      .getOrCreate()

    import spark.implicits._

    // 1. 创建 DataFrame
    val data = Seq(
      ("Alice", 34),
      ("Bob", 45),
      ("Charlie", 28)
    ).toDF("name", "age")

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

    // 3. 读取 Delta 表
    val df = spark.read.format("delta").load("/tmp/delta_table")
    df.show()

    // 4. 追加数据
    val moreData = Seq(("Diana", 31)).toDF("name", "age")
    moreData.write.format("delta").mode("append").save("/tmp/delta_table")

    // 5. 读取历史版本
    val historyDf = spark.read.format("delta")
      .option("versionAsOf", 0)
      .load("/tmp/delta_table")
    historyDf.show()

    spark.stop()
  }
}

运行这个程序，你会看到：

• 第一次 show() 输出 3 条记录
• 追加后第二次读取输出 4 条记录
• 时间旅行读取版本 0，又回到 3 条记录

嗯，到这里你的环境就完全配好了。

3.6 常见问题与避坑指南

我曾经帮一个同事排查环境问题，搞了整整一下午。最后发现是他 pom.xml 里 Delta 版本写成了 2.0.0，但 Spark 是 3.5.0，不兼容。这里我把常见问题列出来：

spark-shell --packages io.delta:delta-core_2.12:3.1.0 \
  --conf "spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension" \
  --conf "spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog"

3.7 验证环境：跑一个完整的 Delta 事务

最后，我建议你跑一个稍微复杂点的场景，验证 ACID 特性是否生效。写两个线程同时往同一个 Delta 表写数据：

import org.apache.spark.sql.SparkSession

object ConcurrentWriteTest {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[2]")
      .appName("ConcurrentWriteTest")
      .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
      .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
      .getOrCreate()

    import spark.implicits._

    // 线程 1：写 1-100
    val t1 = new Thread(() => {
      val data = (1 to 100).map(i => (s"user_$i", i)).toDF("name", "id")
      data.write.format("delta").mode("append").save("/tmp/delta_concurrent")
    })

    // 线程 2：写 101-200
    val t2 = new Thread(() => {
      val data = (101 to 200).map(i => (s"user_$i", i)).toDF("name", "id")
      data.write.format("delta").mode("append").save("/tmp/delta_concurrent")
    })

    t1.start()
    t2.start()
    t1.join()
    t2.join()

    // 验证：总记录数应该是 200
    val count = spark.read.format("delta").load("/tmp/delta_concurrent").count()
    println(s"Total records: $count")  // 应该输出 200

    spark.stop()
  }
}

如果输出是 200，说明 Delta Lake 的并发写入没问题。如果少于 200，那可能是你的环境有问题——不过 Delta Lake 的 ACID 保证不会丢数据，所以大概率是你代码写错了。

好了，环境搭好了，后面我们就可以放心地玩 Delta Lake 的各种特性了。