3、Apache Spark基础：Spark生态介绍，RDD、DataFrame、Dataset概念，SparkSession初始化

好，咱们进入第三章。说实话，Spark 是 Lakehouse 的“发动机”。没有它，湖上那些数据根本跑不起来。我见过不少团队，把数据一股脑丢进对象存储，然后用 Presto 或 Hive 去查——慢得让人抓狂。后来换了 Spark，同样的查询，从分钟级降到了秒级。

这一章，咱们把 Spark 的底裤扒开看看。不绕弯子，直接上干货。

3.1 Spark生态：不只是个计算引擎

很多人以为 Spark 就是个“快一点的 MapReduce”。嗯，也对也不对。它确实比 MapReduce 快很多——内存计算、DAG 调度、向量化执行，这些都是实打实的优化。但 Spark 真正的价值，在于它构建了一个完整的生态。

我习惯把 Spark 生态分成三层：

  • 核心层：Spark Core，负责内存管理、任务调度、容错机制。RDD 就是这层的“一等公民”。
  • API 层：Spark SQL、DataFrame、Dataset、MLlib、GraphX、Structured Streaming。这些是咱们日常打交道的工具。
  • 扩展层：Delta Lake、Apache Hudi、Iceberg、Kafka 集成、JDBC 数据源。Lakehouse 的“湖”和“仓”就是靠这些连起来的。

说白了，Spark 是个“瑞士军刀”。你想做批处理？用 Spark SQL。想做流处理？用 Structured Streaming。想做机器学习？用 MLlib。而且这些组件共享同一个执行引擎，数据不用来回倒腾。

核心观点：Spark 生态的“统一”特性，是 Lakehouse 架构能落地的关键。你不需要为不同场景维护多套引擎，一套 Spark 搞定所有。

我在项目中遇到过最典型的场景：实时流进来的数据，用 Structured Streaming 做轻量清洗，然后直接写入 Delta 表；凌晨再用 Spark SQL 跑全量聚合，生成报表。整个过程，代码风格一致，数据格式统一，运维成本极低。

3.2 RDD、DataFrame、Dataset：三个“兄弟”的区别

这三个概念，是 Spark 入门的“鬼门关”。很多初学者搞不清它们的关系。我试着用最直白的话讲清楚。

3.2.1 RDD：最底层的“砖头”

RDD（Resilient Distributed Dataset）是 Spark 1.x 时代的“老大哥”。它就是一个不可变的、分区的、可以并行计算的数据集合。你可以把它想象成一块块“砖头”，每块砖头分布在不同的机器上。

RDD 的优点：

  • 弹性：某个分区挂了，Spark 能根据血缘关系重新计算，不会丢数据。
  • 类型安全：编译时就能发现类型错误。
  • 灵活：你可以自定义各种复杂的转换操作。

但 RDD 有个致命缺点：性能差。因为它不知道数据的结构（schema），所以没法做优化。比如你写个 filter，Spark 不知道你过滤的是哪一列，只能把整个对象反序列化出来，再判断。这效率能高吗？

避坑指南：我曾经在一个项目里，用 RDD 处理几百 GB 的日志数据。代码写得很“优雅”，但跑起来慢得像蜗牛。后来换成 DataFrame，同样的逻辑，快了 5 倍。所以，除非你需要操作非结构化数据（比如二进制流），否则别碰 RDD。

3.2.2 DataFrame：带“表头”的数据集

DataFrame 是 Spark 1.3 引入的，它本质上就是“带 schema 的 RDD”。你可以把它想象成一张数据库表：有行、有列、每列有名字和类型。

有了 schema，Spark 就能做两件大事：

  • Catalyst 优化器：它会分析你的查询计划，自动做谓词下推、列剪枝、常量折叠等优化。你写同样的逻辑，DataFrame 比 RDD 快很多。
  • Tungsten 执行引擎：它绕过了 Java 对象序列化，直接操作二进制内存。说白了，就是“零拷贝”处理数据。

DataFrame 的 API 是 非类型安全 的。什么意思？你写 df.select("col1")，如果 col1 不存在，编译时不会报错，运行时才抛异常。这算是个小坑。

3.2.3 Dataset：兼顾“性能”和“类型安全”

Dataset 是 Spark 1.6 引入的，它试图融合 RDD 和 DataFrame 的优点。简单说：

  • 在 Java/Scala 中，Dataset 是类型安全的。你定义一个 Dataset<Person>，然后 ds.filter(p -> p.age > 18)，编译时就能检查类型。
  • 在 Python 中，Dataset 退化为 DataFrame。因为 Python 是动态类型语言，没法做编译时类型检查。

所以，如果你用 Scala/Java 写 Spark 作业，我强烈建议用 Dataset。它既有 DataFrame 的性能优化，又有 RDD 的类型安全。Python 用户嘛，老老实实用 DataFrame 就好。

特性 RDD DataFrame Dataset
引入版本 1.0 1.3 1.6
类型安全 是（编译时） 否（运行时） 是（编译时，仅 Java/Scala）
性能 低（无优化） 高（Catalyst + Tungsten） 高（Catalyst + Tungsten）
适用场景 非结构化数据、自定义复杂逻辑 结构化数据、SQL 分析、Python 用户 需要类型安全的复杂 ETL（Java/Scala）
API 风格 函数式（map, filter, reduce） 声明式（select, groupBy, agg） 混合（函数式 + 声明式）

你想想看，为什么 Spark 要搞三个 API？其实是为了适应不同场景。RDD 是“底层能力”，DataFrame 是“生产力”，Dataset 是“两者兼顾”。在实际项目中，我 90% 的时间用 DataFrame，偶尔用 Dataset 做类型安全的复杂转换，几乎不用 RDD。

3.3 SparkSession：一切的起点

在 Spark 2.0 之前，你要初始化 Spark 上下文，得写一堆代码：

// 老版本 Spark 1.x
val conf = new SparkConf().setAppName("MyApp").setMaster("local[*]")
val sc = new SparkContext(conf)
val sqlContext = new SQLContext(sc)
val hiveContext = new HiveContext(sc)  // 如果需要 Hive 支持

是不是很烦？每个组件都要单独初始化，而且容易搞混。

Spark 2.0 之后，SparkSession 统一了这一切。它把 SparkContext、SQLContext、HiveContext、StreamingContext 全部整合到一个入口里。你只需要：

// Scala
import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder()
  .appName("MyLakehouseApp")
  .master("local[*]")
  .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
  .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
  .enableHiveSupport()  // 如果需要 Hive 元数据
  .getOrCreate()

# Python
from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("MyLakehouseApp") \
    .master("local[*]") \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
    .enableHiveSupport() \
    .getOrCreate()

个人习惯：我一般把 SparkSession 的初始化放在一个单独的 SparkUtils 类里，然后整个应用共享同一个实例。这样既避免了重复创建，也方便统一管理配置。

这里有几个关键点要注意：

  • appName：给你的应用起个有意义的名字。在 YARN 或 Kubernetes 上跑的时候，这个名字会出现在资源管理器的 UI 上，方便排查问题。
  • master：本地开发用 local[*]，生产环境用 yarn 或 k8s://。千万别在生产环境用 local 模式，否则你会被运维骂死。
  • config：这里配置了 Delta Lake 的扩展。如果你用 Hudi 或 Iceberg，换成对应的配置就行。
  • enableHiveSupport：如果你需要读取 Hive 表，或者想把元数据存到 Hive Metastore，就开启它。Lakehouse 架构里，Hive Metastore 通常作为“元数据层”存在。

我曾经踩过的坑：有一次，我在生产环境忘了设置 spark.sql.adaptive.enabled=true（自适应查询执行），结果一个简单的 join 操作，因为数据倾斜，跑了 2 小时没跑完。后来加上这个配置，Spark 自动做了分桶和广播，10 分钟搞定。所以，初始化 SparkSession 时，记得把常用的优化配置加上。

初始化完成后，你就可以用 spark 这个对象做各种操作了：

// 读取 CSV 文件
val df = spark.read.option("header", "true").csv("s3://my-bucket/data/")

// 注册临时视图
df.createOrReplaceTempView("my_table")

// 用 SQL 查询
val result = spark.sql("SELECT category, COUNT(*) FROM my_table GROUP BY category")

// 写入 Delta 表
result.write.format("delta").mode("overwrite").save("/lakehouse/silver/agg_category")

你看，是不是很简洁？所有操作都通过 spark 这个入口完成，不需要再纠结用 sqlContext 还是 hiveContext。

3.4 本章小结：一张图看懂 Spark 基础

说了这么多，咱们用一张图把 Spark 生态、三个 API 的关系、以及 SparkSession 的定位串起来。

Spark 基础架构与 API 关系 SparkSession（统一入口） RDD（底层能力） DataFrame（生产力） Dataset（兼顾两者） Spark Core（内存管理、任务调度、容错） Catalyst 优化器 + Tungsten 执行引擎 数据源：Delta / Hudi / Iceberg / Kafka / JDBC 说明：SparkSession 统一了所有 API 的入口；DataFrame/Dataset 底层依赖 Catalyst + Tungsten 做优化； RDD 则直接跑在 Spark Core 上，没有优化层。Lakehouse 的数据源通过 Spark 的 DataSource API 接入。

嗯，这张图基本把本章的核心内容都涵盖了。你记住一个原则：能用 DataFrame 就别用 RDD，能用 Dataset 就别裸写 map。Spark 的优化器不是吃干饭的，你给它越多结构信息，它就越能帮你省时间。

← 上一章 📖 返回目录 下一章 →