4. 批处理引擎：Apache Spark 核心概念、RDD 与 DataFrame、Spark SQL 与 Hive 集成、性能调优

好，咱们进入第四章。这一章我打算把 Spark 这个批处理引擎的核心东西给你捋一遍。说实话，Spark 在流批一体架构里太重要了，你后面做实时离线融合，底层基本都离不开它。我当年刚接触 Spark 时，也被 RDD、DataFrame、DataSet 这些概念绕得头晕。后来踩了不少坑，才慢慢摸清楚它们各自的门道。

咱们今天不讲虚的，直接上干货。从核心概念开始，一路聊到性能调优。你跟着我的节奏走，保证能把这些东西吃透。

4.1 Spark 核心概念：你必须要懂的底层逻辑

Spark 的核心，说白了就是一个基于内存的分布式计算框架。它跟 Hadoop MapReduce 最大的区别是什么？我个人的理解是：MapReduce 每个 stage 都要落盘，而 Spark 尽量在内存里搞定。这就好比一个快递员，MapReduce 是每送一个包裹就回仓库一趟，Spark 是装一车包裹一次性送完。速度自然快得多。

Spark 的架构里，有几个角色你得记住：

  • Driver：整个应用的“大脑”。它负责解析你的代码，生成执行计划，然后调度任务。我在项目中遇到过 Driver 内存不够导致任务失败的情况，那叫一个惨。所以 Driver 的内存配置一定要留足。
  • Executor：真正干活的“工人”。每个 Executor 运行在集群的节点上，负责执行具体的 task，并存储计算数据。Executor 的数量和内存大小，直接决定了你的作业能跑多快。
  • Cluster Manager：资源调度器。可以是 Standalone、YARN 或者 Mesos。我个人习惯用 YARN，因为它跟 Hadoop 生态集成得最好。

核心要点： Spark 应用 = Driver + Executors。Driver 负责任务调度，Executors 负责数据计算。资源分配是调优的第一步。

4.2 RDD 与 DataFrame：从底层到上层的进化

RDD（弹性分布式数据集）是 Spark 最底层的抽象。它就像一个分布式的集合，你可以对它做各种转换操作（map、filter、flatMap 等）。但 RDD 有个问题——它不知道数据的结构。它只把数据当成一堆 Java/Python 对象。这就导致 Spark 没法针对数据做优化。

举个例子，你用 RDD 做过滤，Spark 只能傻乎乎地遍历所有数据。但如果你用 DataFrame，Spark 就知道你有几列、每列是什么类型，它就能用 Catalyst 优化器生成更高效的执行计划。

DataFrame 是什么？你可以把它想象成一张数据库表。它有 schema（列名和类型），有行。DataFrame 的 API 更高级，写起来也更简洁。我刚开始用 Spark 时，习惯写 RDD 代码，后来发现 DataFrame 不仅代码少，跑起来还快。为什么？因为 DataFrame 背后有 Catalyst 优化器帮你做各种优化，比如谓词下推、列剪枝。

我的建议： 能用 DataFrame 就别用 RDD。除非你需要做一些 RDD 特有的操作（比如自定义分区），否则 DataFrame 是更好的选择。性能差距可能达到 2-5 倍。

咱们看个代码对比：

// RDD 方式：手动处理每一行
val rdd = sc.textFile("hdfs://data/sales.csv")
val filteredRDD = rdd.filter(line => line.split(",")(3).toDouble > 1000)

// DataFrame 方式：声明式操作
val df = spark.read.option("header", "true").csv("hdfs://data/sales.csv")
val filteredDF = df.filter($"amount" > 1000)

你看，DataFrame 的代码更直观，而且 Spark 会自动优化过滤条件。RDD 的代码虽然也能跑，但优化空间小得多。

4.3 Spark SQL 与 Hive 集成：打通数据仓库

Spark SQL 是 Spark 生态里最实用的组件之一。它让你可以用 SQL 语句操作 DataFrame。这对于熟悉 SQL 的数据分析师来说，简直是福音。

但更关键的是 Spark SQL 与 Hive 的集成。Hive 是 Hadoop 生态里的数据仓库工具，很多公司都用它存储历史数据。Spark SQL 可以直接读取 Hive 表，也能把计算结果写回 Hive。这就实现了批处理场景下的数据打通。

我记得有一次，我们需要对 Hive 里存储的半年用户行为数据做分析。如果用 Hive 跑，一个复杂的 join 可能要跑几个小时。但换成 Spark SQL，同样的逻辑，半小时就搞定了。为什么？因为 Spark 的内存计算能力比 Hive 的 MapReduce 引擎快得多。

配置 Spark SQL 与 Hive 集成，你需要做这几步：

  1. 把 hive-site.xml 复制到 Spark 的 conf 目录下。
  2. 启动 Spark 时加上 --conf spark.sql.catalogImplementation=hive。
  3. 在代码里用 spark.sql("SELECT * FROM hive_table") 直接查询。

注意： 我曾经遇到过 Spark 和 Hive 的元数据版本不兼容的问题。建议你确保 Spark 和 Hive 的版本匹配，否则可能连表都读不出来。

下面这张图展示了 Spark SQL 与 Hive 集成的整体架构：

Spark SQL 与 Hive 集成架构图 用户 SQL / DataFrame API Catalyst 优化器 Spark SQL 执行引擎 Hive Metastore（元数据） HDFS / Hive 表数据 计算结果（DataFrame） 集成关键点 • 共享 Hive Metastore • 直接读取 Hive 表 • 兼容 Hive SQL 语法 • 性能比 Hive MR 快 5-10 倍

4.4 Spark 性能调优：实战中的血泪经验

性能调优这块，我踩过的坑最多。说实话，Spark 调优没有银弹，但有几个方向你只要抓住了，大部分问题都能解决。

4.4.1 资源分配：别让 Executor 饿着

资源分配是调优的第一步。你想想看，如果 Executor 的内存太小，数据频繁溢写到磁盘，性能肯定上不去。我一般这样配置：

参数 建议值 说明
spark.executor.memory 4g - 8g 每个 Executor 的内存，不要超过物理内存的 80%
spark.executor.cores 2 - 4 每个 Executor 的 CPU 核数，太多会导致 I/O 竞争
spark.driver.memory 2g - 4g Driver 内存，如果做 collect 操作要加大
spark.sql.shuffle.partitions 200 - 500 shuffle 后的分区数，根据数据量调整

避坑指南： 我曾经把 spark.executor.memory 设得太大，结果 YARN 直接 kill 了 Executor。后来才知道，要给系统预留内存。建议 Executor 内存不超过节点物理内存的 80%。

4.4.2 数据倾斜：最头疼的问题

数据倾斜是 Spark 作业里最常见的性能杀手。说白了，就是某个 key 的数据量特别大，导致一个 task 要处理大部分数据，其他 task 早早跑完等着它。

怎么解决？我常用的方法有：

  • 加盐（Salting）：给倾斜的 key 加上随机前缀，把数据分散到多个分区。处理完后再去掉前缀。
  • 广播小表：如果一个大表 join 一个小表，把小表广播到每个 Executor，避免 shuffle。
  • 调整分区数：增加 spark.sql.shuffle.partitions，让每个分区处理的数据更少。

举个例子，加盐的代码大概是这样：

// 假设 user_id 为 "hot_user" 的数据严重倾斜
val saltedDF = df.withColumn("salted_key", 
  when($"user_id" === "hot_user", 
    concat($"user_id", lit("_"), (rand() * 10).cast("int")))
  .otherwise($"user_id"))

// 按加盐后的 key 做聚合
val result = saltedDF.groupBy("salted_key").agg(...)

// 去掉盐前缀再做最终聚合
val finalResult = result.withColumn("real_key", 
  regexp_replace($"salted_key", "_\\d+$", ""))
  .groupBy("real_key").agg(...)

4.4.3 序列化与内存管理

Spark 默认使用 Java 序列化，但性能很差。我建议你用 Kryo 序列化，速度能快 10 倍。

sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
sparkConf.registerKryoClasses(Array(classOf[MyClass]))

内存管理方面，Spark 把内存分为执行内存和存储内存。执行内存用于计算（shuffle、join），存储内存用于缓存数据。默认比例是 0.5:0.5。如果你的作业有大量缓存操作，可以调高存储内存比例。

注意： 不要盲目缓存数据。我见过有人把几百 GB 的数据全 cache 到内存，结果 Executor 直接 OOM。缓存前先想想，这些数据真的会被多次使用吗？

4.4.4 动态资源分配

这个功能我强烈推荐开启。它能让 Spark 根据作业的负载动态调整 Executor 数量。空闲时释放资源，繁忙时申请资源。对于流批一体的场景特别有用。

spark.dynamicAllocation.enabled=true
spark.dynamicAllocation.minExecutors=1
spark.dynamicAllocation.maxExecutors=50
spark.dynamicAllocation.initialExecutors=5

嗯，到这里 Spark 的核心内容就差不多了。你把这些概念和调优方法吃透了，后面做流批一体融合时，就能更从容地应对各种性能问题。记住，Spark 调优不是一蹴而就的，需要你在实际项目中不断摸索和总结。

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