共计 2388 个字符，预计需要花费 6 分钟才能阅读完成。
近期接手了不少大数据表任务调度补数据的工作, 补数时发现资源消耗异常的大且运行速度却不怎么给力.
发现根本原因在于 sparkSQL 配置有诸多问题, 解决后总结出来就当抛砖引玉了.
具体现象
- 内存 CPU 比例失调 一个 Spark 任务消耗 120(executor)*4G = 480G 内存仅仅使用 120 个 core. 几个 SprakSQL 任务就将整个系统资源吃光.
- 设置超过 40 个 executor, 但未指定分区数, 导致多数 executor 空闲.
原因分析
SparkSQL 配置时 Core 与内存比例不恰当
没有指定 executor 核心数
未进行其他配置参数优化
解决办法
- 在配置 SparkSQL 任务时指定 executor 核心数 建议为 4
(同一 executor[进程] 内内存共享, 当数据倾斜时, 使用相同核心数与内存量的两个任务,executor 总量少 的任务不容易 OOM, 因为单核心最大可用内存大. 但是并非越大越好, 因为单个 exector 最大 core 受服务器剩余 core 数量限制,过大的 core数量可能导致资源分配不足) - 设置 spark.default.parallelism=600 每个 stage 的默认 task 数量
(计算公式为num-executors * executor-cores 系统默认值分区为 40, 这是导致 executor 并行度上不去的罪魁祸首, 之所以这样计算是为了尽量避免计算 最慢的 task决定整个 stage 的时间, 将其设置为总核心的 2 - 3 倍, 让运行快的 task 可以继续领取任务计算直至全部任务计算完毕) - 开启 spark.sql.auto.repartition=true 自动重新分区
(每个stage[阶段] 运行时分区并不尽相同, 使用此配置可优化计算后分区数, 避免分区数过大导致单个分区数据量过少, 每个 task 运算分区数据时时间过短, 从而导致 task 频繁调度消耗过多时间) - 设置 spark.sql.shuffle.partitions=400 提高 shuffle 并行度
(shuffle read task 的并行度) - 设置 spark.shuffle.service.enabled=true 提升 shuffle 效率 –! 并未测试
(Executor 进程除了运行 task 也要进行写 shuffle 数据, 当 Executor 进程任务过重时, 导致 GC 不能为其他 Executor 提供 shuffle 数据时将会影响效率. 此服务开启时代替 Executor 来抓取 shuffle 数据)
- 在配置 SparkSQL 任务时指定 executor 核心数 建议为 4
前后资源配置对比
| 类型 | 内存数量 | cpu 核心数量 | executor 数量 | executor 内存 | 单核心内存 |
|---|---|---|---|---|---|
| 系统资源总量 | 7168G | 3500 | – | – | 2G |
| 目前一个任务 | 480G | 120 | 120 | 4G | 4G |
| 优化后 | 480G | 240 | 60 | 8G | 2G |
以下为 SparkSQL 调优相关设置
以下列表中动态资源分配相关不建议使用
//1. 下列 Hive 参数对 Spark 同样起作用。
set hive.exec.dynamic.partition=true; // 是否允许动态生成分区
set hive.exec.dynamic.partition.mode=nonstrict; // 是否容忍指定分区全部动态生成
set hive.exec.max.dynamic.partitions = 100; // 动态生成的最多分区数
//2. 运行行为
set spark.sql.autoBroadcastJoinThreshold; // 大表 JOIN 小表，小表做广播的阈值
set spark.dynamicAllocation.enabled; // 开启动态资源分配
set spark.dynamicAllocation.maxExecutors; // 开启动态资源分配后，最多可分配的 Executor 数
set spark.dynamicAllocation.minExecutors; // 开启动态资源分配后，最少可分配的 Executor 数
set spark.sql.shuffle.partitions; // 需要 shuffle 是 mapper 端写出的 partition 个数
set spark.sql.adaptive.enabled; // 是否开启调整 partition 功能，如果开启，spark.sql.shuffle.partitions 设置的 partition 可能会被合并到一个 reducer 里运行
set spark.sql.adaptive.shuffle.targetPostShuffleInputSize; // 开启 spark.sql.adaptive.enabled 后，两个 partition 的和低于该阈值会合并到一个 reducer
set spark.sql.adaptive.minNumPostShufflePartitions; // 开启 spark.sql.adaptive.enabled 后，最小的分区数
set spark.Hadoop.mapreduce.input.fileinputformat.split.maxsize; // 当几个 stripe 的大小大于该值时，会合并到一个 task 中处理
//3.executor 能力
set spark.executor.memory; // executor 用于缓存数据、代码执行的堆内存以及 JVM 运行时需要的内存
set spark.yarn.executor.memoryOverhead; //Spark 运行还需要一些堆外内存，直接向系统申请，如数据传输时的 netty 等。
set spark.sql.windowExec.buffer.spill.threshold; // 当用户的 SQL 中包含窗口函数时，并不会把一个窗口中的所有数据全部读进内存，而是维护一个缓存池，当池中的数据条数大于该参数表示的阈值时，spark 将数据写到磁盘
set spark.executor.cores; // 单个 executor 上可以同时运行的 task 数
：
