3、环境搭建：安装配置Spark与Delta Lake、创建第一个Delta表

3.1 核心原理：Delta Lake 与 Spark 的运行时关系

在动手安装之前，必须先理解 Delta Lake 并非独立运行的存储引擎，而是构建在 Apache Spark 之上的事务性存储层。它通过一组 Spark 的扩展接口（如 SparkSessionExtensions）和 Catalyst 优化器规则，将 ACID 事务、时间旅行、Schema 强制等能力注入 Spark 的 DataFrame 与 SQL 执行流程中。因此，安装配置的核心不是“启动一个 Delta 服务”，而是让 Spark 在启动时能够加载 Delta Lake 的 JAR 包，并正确注册其扩展。

从架构上看，Delta Lake 的写入路径会经过 Spark 的分布式执行引擎，将数据文件（Parquet 格式）和事务日志（_delta_log 目录下的 JSON 文件）原子性地提交到存储系统（本地文件系统、HDFS 或 S3）。这意味着：Spark 是计算引擎，Delta Lake 是元数据与事务管理层，两者通过 JVM 进程内的类加载机制协同工作。理解这一点，就能明白为什么版本匹配至关重要——Delta Lake 0.6.x 对应 Spark 2.4，而 Delta Lake 2.x 则要求 Spark 3.x。

3.2 方法流程：从零搭建 Spark + Delta Lake 开发环境

本节以 macOS / Linux 环境 + Spark 3.3.2 + Delta Lake 2.3.0 为例，演示完整的安装配置流程。整个过程分为四个步骤：准备 Java 与 Scala 环境、下载并配置 Spark、启动 Spark Shell 时加载 Delta Lake 包、验证安装是否成功。

步骤一：检查基础环境

Delta Lake 运行在 JVM 之上，因此 Java 8/11/17 均可（Spark 3.3 官方推荐 Java 8/11）。使用以下命令确认环境：

# 检查 Java 版本（需为 8 或 11）
java -version

# 检查 Scala（Spark 自带，无需单独安装）
# 检查 Python（可选，用于 PySpark）
python3 --version

步骤二：下载并配置 Spark

从 Apache 官网下载预编译的 Spark 发行版（选择 Pre-built for Apache Hadoop 3.3 或类似版本），解压后配置环境变量：

# 下载 Spark 3.3.2（含 Hadoop 3.3）
wget https://archive.apache.org/dist/spark/spark-3.3.2/spark-3.3.2-bin-hadoop3.tgz
tar -xzf spark-3.3.2-bin-hadoop3.tgz
sudo mv spark-3.3.2-bin-hadoop3 /opt/spark

# 配置环境变量（写入 ~/.bashrc 或 ~/.zshrc）
export SPARK_HOME=/opt/spark
export PATH=$SPARK_HOME/bin:$PATH
export PYSPARK_PYTHON=python3

# 使配置生效
source ~/.bashrc

步骤三：启动 Spark Shell 并加载 Delta Lake

Delta Lake 的 JAR 包可以通过 Maven 坐标自动下载。使用 --packages 参数指定 Delta Lake 版本，Spark 会自动从 Maven 中央仓库拉取依赖：

# 启动 Scala Spark Shell 并加载 Delta Lake 2.3.0
spark-shell --packages io.delta:delta-core_2.12:2.3.0

# 如果使用 PySpark，则执行：
# pyspark --packages io.delta:delta-core_2.12:2.3.0

启动成功后，控制台会显示 Spark 版本和 Delta Lake 的加载日志。此时在 Scala Shell 中执行以下验证代码：

// 验证 Delta Lake 是否可用
import io.delta.tables._
val df = spark.range(10)
df.write.format("delta").save("/tmp/delta_verify")
val readBack = spark.read.format("delta").load("/tmp/delta_verify")
readBack.show()

如果能够正常输出 0~9 的数字，说明 Delta Lake 已成功集成到 Spark 中。注意：写入路径必须使用 format("delta") 显式指定，否则 Spark 默认使用 Parquet 格式，不会产生事务日志。

3.3 案例实战：创建第一个 Delta 表并验证事务特性

下面通过一个完整的案例，演示如何创建 Delta 表、写入数据、读取数据，并验证 ACID 事务的核心特性——原子性与时间旅行。我们使用 Spark SQL 的 USING DELTA 语法来建表，这是最直观的方式。

3.3.1 使用 SQL 创建 Delta 表

// 在 Spark Shell 中执行 SQL
spark.sql("CREATE DATABASE IF NOT EXISTS demo")
spark.sql("USE demo")

// 创建 Delta 表（本质是创建托管表，数据存储在 Spark 仓库目录下）
spark.sql("""
  CREATE TABLE IF NOT EXISTS user_events (
    user_id INT,
    event_time TIMESTAMP,
    event_type STRING,
    amount DOUBLE
  )
  USING DELTA
  PARTITIONED BY (event_date DATE)
""")

// 插入第一批数据
spark.sql("""
  INSERT INTO user_events VALUES
    (1, '2024-01-15 10:30:00', 'purchase', 99.9, DATE '2024-01-15'),
    (2, '2024-01-15 11:00:00', 'click', 0.0, DATE '2024-01-15'),
    (3, '2024-01-16 09:15:00', 'purchase', 199.0, DATE '2024-01-16')
""")

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

执行后，可以在 Spark 的仓库目录（默认是 spark-warehouse）下看到 user_events 文件夹，其中包含 Parquet 数据文件和 _delta_log 目录。事务日志中记录了每次操作的版本号（如 00000000000000000000.json）。

3.3.2 验证时间旅行与数据回滚

Delta Lake 最强大的特性之一是时间旅行。假设我们执行了一次误操作，删除了部分数据：

// 模拟误操作：删除 user_id = 2 的记录
spark.sql("DELETE FROM user_events WHERE user_id = 2")

// 此时表内只剩 2 条记录
spark.sql("SELECT * FROM user_events").show()

// 使用时间旅行恢复到删除前的版本（版本 0）
val df_restored = spark.read.format("delta")
  .option("versionAsOf", "0")
  .table("demo.user_events")
df_restored.show()

通过 versionAsOf 选项，我们可以读取任意历史版本的数据。Delta 表的每次写操作（INSERT、DELETE、UPDATE、MERGE）都会生成一个新的版本，版本号从 0 开始递增。下表总结了常用操作与版本的关系：

3.4 常见误区与复盘：环境搭建中的“坑”

根据大量实战经验，环境搭建阶段最容易出现以下四类问题。理解这些误区能帮助你快速定位故障，避免浪费时间。

spark.jars.packages  io.delta:delta-core_2.12:2.3.0
spark.sql.extensions  io.delta.sql.DeltaSparkSessionExtension
spark.sql.catalog.spark_catalog  org.apache.spark.sql.delta.catalog.DeltaCatalog