2. 环境搭建：Spark + Delta Lake 本地开发环境配置，PySpark 与 Delta Lake 集成

说实话，很多初学者在 Delta Lake 上栽跟头，不是因为概念多难，而是环境没配好。我见过有人花了两天时间折腾依赖冲突，最后发现是 Scala 版本不对。所以这一章，咱们把环境彻底捋清楚。

我会带你从零搭建一套能跑 Delta Lake 实时特征工程的本地环境。包括 Spark 安装、Delta Lake 依赖引入、PySpark 与 Delta Lake 的集成，以及一个完整的端到端验证。嗯，跟着我的节奏来，半小时内你就能在本地跑通第一个 Delta Lake 写读流程。

2.1 整体架构概览

先看一张图，搞清楚我们今天要搭的这套环境长什么样。

这张图其实就三层：你的 PySpark 代码 → SparkSession（带着 Delta Lake 插件）→ 本地文件系统里的 Delta 表。中间那个 _delta_log 目录是 Delta Lake 的灵魂，所有事务记录都写在那里。

2.2 前置条件检查

动手之前，先确认你的机器上已经装好了这些基础工具。我个人习惯用 Java 8，虽然 Java 11 也能跑，但 Spark 3.x 对 Java 8 的兼容性最好，踩坑最少。

2.3 安装 Apache Spark

Spark 的安装其实不复杂，但有几个细节要注意。我建议用预编译的二进制包，省去自己编译的麻烦。

2.3.1 下载与解压

# 下载 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 -C /opt/
sudo mv /opt/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

你想想看，为什么我要强调 PYSPARK_PYTHON 这个变量？因为 PySpark 默认会找系统里的 python 命令，如果你的机器上同时有 Python 2 和 Python 3，不指定的话很容易跑错解释器。我刚开始用 PySpark 时就吃过这个亏，脚本里明明写了 Python 3 的语法，结果调的是 Python 2 的解释器，报错报得莫名其妙。

2.3.2 验证安装

# 启动 Spark Shell 验证
spark-shell --version

# 输出示例：
# Welcome to Spark 3.3.2
# Using Scala 2.12.17

看到版本号出来，说明 Spark 装好了。接下来我们装 Delta Lake。

2.4 配置 Delta Lake 依赖

Delta Lake 不是 Spark 自带的，需要额外引入。有两种方式：一种是用 --packages 参数自动下载，另一种是手动下载 jar 包。我推荐第一种，省事。

2.4.1 使用 --packages 参数（推荐）

# 启动 PySpark 时直接引入 Delta Lake
pyspark --packages io.delta:delta-core_2.12:2.3.0 \
         --conf "spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension" \
         --conf "spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog"

这里有三件事要说明白：

• delta-core_2.12:2.3.0：2.12 是 Scala 版本，要和你的 Spark 编译版本一致。Spark 3.3.2 默认用 Scala 2.12。
• spark.sql.extensions：告诉 Spark 加载 Delta Lake 的 SQL 扩展，这样你才能用 Delta 相关的 SQL 语法。
• spark.sql.catalog.spark_catalog：把默认的 Spark catalog 替换成 Delta Catalog，这样表操作才能走 Delta 的事务日志。

2.4.2 手动下载 jar 包（离线环境）

如果你在的公司网络受限，或者你像我一样喜欢把依赖本地化，可以手动下载 jar 包放到 Spark 的 jars 目录下。

# 下载 Delta Lake 核心 jar
wget https://repo1.maven.org/maven2/io/delta/delta-core_2.12/2.3.0/delta-core_2.12-2.3.0.jar

# 下载 Delta Storage 依赖
wget https://repo1.maven.org/maven2/io/delta/delta-storage/2.3.0/delta-storage-2.3.0.jar

# 复制到 Spark jars 目录
cp delta-*.jar $SPARK_HOME/jars/

然后启动 PySpark 时就不需要 --packages 了，但扩展和 catalog 配置还是要加。

2.5 PySpark 与 Delta Lake 集成

环境搭好了，我们来写第一段代码。这段代码会创建一个 Delta 表，写入数据，再读出来。说白了就是验证整个链路通不通。

2.5.1 创建 SparkSession

from pyspark.sql import SparkSession

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()

print("SparkSession 创建成功！")
print(f"Delta Lake 版本: {spark.conf.get('spark.sql.extensions')}")

这里 local[*] 表示用本地模式运行，* 代表使用所有可用的 CPU 核心。开发环境这样够了，生产环境当然要用集群模式。

2.5.2 写入数据到 Delta 表

# 创建一个简单的 DataFrame
data = [("user_001", "click", 1680000000),
        ("user_002", "view", 1680000001),
        ("user_003", "purchase", 1680000002)]

columns = ["user_id", "event_type", "timestamp"]
df = spark.createDataFrame(data, columns)

# 写入为 Delta 表（本地文件系统）
df.write.format("delta").mode("overwrite").save("/tmp/delta_demo/events")

print("数据已写入 Delta 表！")

写入完成后，你去 /tmp/delta_demo/events 目录下看看，会发现多了一个 _delta_log 文件夹。里面就是事务日志，记录了这次写入操作的元数据。

2.5.3 读取 Delta 表

# 读取 Delta 表
df_read = spark.read.format("delta").load("/tmp/delta_demo/events")
df_read.show()

# 输出：
# +-------+----------+----------+
# |user_id|event_type|timestamp |
# +-------+----------+----------+
# |user_01|click     |1680000000|
# |user_02|view      |1680000001|
# |user_03|purchase  |1680000002|
# +-------+----------+----------+

看到数据出来了，说明你的环境已经跑通了。就这么简单。

2.6 验证 Delta Lake 核心特性

光能读写还不够，我们得验证 Delta Lake 的几个核心特性在本地环境是否正常工作。我挑三个最常用的来测：时间旅行、Schema 演进、ACID 事务。

2.6.1 时间旅行

# 先写入第二批数据
data_v2 = [("user_004", "click", 1680000010)]
df_v2 = spark.createDataFrame(data_v2, columns)
df_v2.write.format("delta").mode("append").save("/tmp/delta_demo/events")

# 查看当前数据（4条）
print("当前数据：")
spark.read.format("delta").load("/tmp/delta_demo/events").show()

# 查看历史版本
print("历史版本：")
spark.sql("DESCRIBE HISTORY delta.`/tmp/delta_demo/events`").select("version", "timestamp", "operation").show()

# 读取第一个版本（时间旅行）
print("版本 0 的数据：")
spark.read.format("delta").option("versionAsOf", "0").load("/tmp/delta_demo/events").show()

时间旅行是 Delta Lake 的杀手锏。你想想看，传统数据湖里数据覆盖了就找不回来了，但 Delta Lake 可以随时回溯到任意历史版本。我在做实时特征回溯时经常用这个功能——比如要重新计算某个时间窗口的特征，直接读那个时间点的快照就行，不用重新跑整个 pipeline。

2.6.2 Schema 演进

# 新增一列
data_v3 = [("user_005", "click", 1680000020, "chrome")]
columns_v3 = ["user_id", "event_type", "timestamp", "browser"]
df_v3 = spark.createDataFrame(data_v3, columns_v3)

# 允许 Schema 演进
df_v3.write.format("delta").mode("append") \
    .option("mergeSchema", "true") \
    .save("/tmp/delta_demo/events")

# 查看新 Schema
spark.read.format("delta").load("/tmp/delta_demo/events").printSchema()

2.6.3 ACID 事务验证

# 模拟并发写入（在同一个 SparkSession 中）
from pyspark.sql import functions as F

# 事务 A：写入一批数据
try:
    df_a = spark.range(100).withColumn("batch", F.lit("A"))
    df_a.write.format("delta").mode("append").save("/tmp/delta_demo/acid_test")
    print("事务 A 成功")
except Exception as e:
    print(f"事务 A 失败: {e}")

# 事务 B：同时写入（模拟并发）
try:
    df_b = spark.range(100).withColumn("batch", F.lit("B"))
    df_b.write.format("delta").mode("append").save("/tmp/delta_demo/acid_test")
    print("事务 B 成功")
except Exception as e:
    print(f"事务 B 失败: {e}")

# 最终数据量应该是 200 条
count = spark.read.format("delta").load("/tmp/delta_demo/acid_test").count()
print(f"最终数据量: {count}")  # 输出 200

如果有一个事务失败，Delta Lake 会自动回滚，不会出现部分写入的情况。这就是 ACID 事务的价值。

2.7 常见问题与避坑指南

环境搭建过程中，有几个问题我几乎每次培训都会遇到。列出来给你参考。

2.8 开发环境推荐配置

最后，分享一套我用了很久的开发环境配置。这套配置兼顾了开发效率和调试体验。

我个人习惯用 VS Code 写 PySpark 脚本，然后在终端里用 spark-submit 提交。调试的时候开一个 Jupyter Notebook，边写边跑，效率很高。你找到自己顺手的方式就行。