HealeJean的梦想博客

一个高级软件开发工程师的成长之路

大数据_Spark实战篇

发表于 2025-03-25

前言

Github：https://github.com/HealerJean

博客：http://blog.healerjean.com

一、任务执行监控

1、`Spark Jobs`

1）字段解释

a、`Job Id`

含义：每个 Job 的唯一编号。
特点：
- 从 0 开始递增；
- 每个 SQL 或 RDD 操作（如 count()、saveAsTable()）都会触发一个 Job；
- 注意：一个复杂的 SQL 可能生成多个 Job（比如先 JOIN 再 GROUP BY）。

b、`Description`

含义：该 Job 对应的操作描述。
常见来源：
- sql at NativeMethodAccessorImpl.java:0 → 表示这是一个 SQL 查询（通常是 Hive/Spark SQL）；
- run at ThreadPoolExecutor.java:1149 → 表示是一个 RDD 操作（如 map, filter, reduce 等）；
- Merge Small File → Hive 写入后自动合并小文件（Hive on Spark 特有）。
提示：
- 点击链接可跳转到具体 Stage 或 Task 详情；
- 如果是 SQL，通常会显示部分 SQL 片段（但不会完整打印）。

c、`Submitted`

含义：该 Job 被提交到集群的时间。
格式：YYYY/MM/DD HH:MM:SS
用途：
- 判断任务延迟；
- 分析调度等待时间（比如前面 Job 完成慢导致后续提交晚）；
- 查看是否有“卡住”的 Job（长时间未开始）。

d、`Duration`

Duration 不等于“实际计算时间”，还包括调度、资源申请等开销。

含义：该 Job 执行所花费的时间（从提交到完成）。
单位：秒（s）、分钟（min）
关键点：
- 反映执行效率；
- 若某个 Job 过长（如 >10min），可能是瓶颈；
- 注意区分“运行时间”与“等待时间”。

e、`Stages`: `Succeeded`/`Total`

含义：该 Job 包含的 Stage 数量，以及成功 vs 总数。
格式：成功数 / 总数（例如 1/1）
附加信息：
- (1 failed)：表示有一个 Stage 失败；
- (1 skipped)：表示有一个 Stage 被跳过（如数据为空）；
- 如果是 2/2 (1 failed)，说明有两个 Stage，其中一个失败了。

f、`Tasks` (`for all stages`): `Succeeded/Total`

含义：该 Job 中所有 Stage 的总 Task 数量，以及成功 vs 总数。
格式：成功数 / 总数（例如 6000/6000）
附加信息：
- (11 failed)：有 11 个 Task 失败；
- (4304 skipped)：有 4304 个 Task 被跳过（通常是因数据为空）；
- (4 killed: Stage finished)：4 个 Task 被杀掉，但因为 Stage 已完成，所以不影响结果。
重点观察：
- 如果 failed 数量多 → 需要排查原因；
- 如果 skipped 很高 → 数据可能为空或条件不满足；
- killed 通常是正常现象（如 Stage 结束后清理）。

2）实战技巧

场景	应对策略
`Job` 失败	查 `Stages` 页面 → 找 `Failed Stage` → 查 `stderr`
`Task` 失败多	查 `Executors` 页面 → 找 `Failed Tasks` 的 `Executor` → 查 stderr
`Task` 被跳过	检查是否数据为空，或过滤条件太严格
`Job` 耗时长	查 `Stages 页面 → 找 Duration 最长的` Stage `→ 分析 Shuffle 和 Input
有 `skip`	可能是动态分区写入时部分分区为空

2、监控： `Stages`

列名	含义	关键用途
Stage Id	阶段编号	识别执行顺序
Description	操作描述	定位业务逻辑
Submitted	提交时间	分析调度延迟
Duration	执行时长	发现性能瓶颈
Tasks: Succeeded/Total	成功任务数 / 总任务数	判断执行成功率
Input	输入数据量	评估数据规模
Output	输出数据量	判断处理效果
Shuffle Read	读取的 Shuffle 数据	分析网络负载
Shuffle Write	写入的 Shuffle 数据	分析磁盘/网络压力
Failure Reason	失败原因	排查问题根源

1）字段解释

a、`Stage Id`

含义：该 Stage 的唯一编号。
特点：
- 从 0 开始递增；
- 通常按依赖顺序执行；
- 如果有重试，会显示 (retry N)，如 4 (retry 1) 表示第 4 阶段重试了 1 次。
示例：Stage 0 是第一个执行的阶段，Stage 5 是最后一个完成的。

b、`Description`

含义：该 Stage 对应的操作描述。
常见来源：
- sql at NativeMethodAccessorImpl.java:0 → 通常是 SQL 查询触发的 Shuffle 或 Join；
- run at ThreadPoolExecutor.java:1149 → 一般是 RDD 操作（如 map、filter 等）；
- Merge Small File → Hive 写入后自动合并小文件（Hive on Spark 特有）。
提示：点击链接可跳转到具体代码行或操作细节。

c、`Submitted`

含义：该 Stage 被提交到集群的时间。
格式：YYYY/MM/DD HH:MM:SS
用途：
- 判断任务延迟；
- 分析调度等待时间（比如前面 Stage 完成慢导致后续提交晚）。

d、`Duration`

含义：该 Stage 执行所花费的时间（从开始到结束）。
单位：秒（s）、分钟（min）等。
关键点：
- 反映执行效率；
- 若某个 Stage 过长（如 >10min），可能是瓶颈；
- 注意区分“运行时间”与“等待时间”。

e、`Tasks: Succeeded/Total`

含义：该 Stage 中成功 vs 总共的任务数。
格式：成功数 / 总数（例如 6000/6000）
附加信息：
- 如 3 killed: Stage finished → 3 个 Task 被杀掉，但因为 Stage 已完成，所以不影响结果；
- 11 failed → 有 11 个 Task 失败（需要重点排查）；
- killed 不一定代表错误，可能是调度器主动终止（如超时、资源不足）。
注意： 失败任务越多，越要关注！

f、`Input`

含义：该 Stage 读取的数据量（来自 HDFS、S3、数据库等）。
单位：GB / MB
用途：
- 判断输入数据规模；
- 如果 Input 很大但处理快，说明并行度高；
- 如果 Input 小但耗时长，可能有计算瓶颈。

g、 `Output`

含义：该 Stage 输出的数据量（写入磁盘或传递给下一个 Stage）。
单位：GB / MB
注意：
- Output ≠ Input，尤其是经过聚合、过滤后；
- 若 Output 显著小于 Input，说明有大量数据被过滤掉（合理）；
- 若 Output 接近 Input，可能没有有效压缩或去重。

h、`Shuffle Read`

含义：该 Stage 从其他 Executor 上拉取的 Shuffle 数据量。
场景：
- 出现在 reduce 类操作中（如 groupByKey, join, reduceByKey）；
- 是 网络 I/O 的主要开销；
典型值：
- 小型作业：几十 MB；
- 大型作业：几 GB ~ 几百 GB；
优化方向：
- 减少 Shuffle 数据量（如预聚合、广播小表）；
- 优化分区策略；
- 启用 AQE 自动优化。

i、`Shuffle Write`

含义：该 Stage 向其他 Executor 写入的 Shuffle 数据量。
场景：
- 发生在 map 端的 Shuffle（如 join 的左表写入）；
- 是 磁盘 I/O 和网络发送的主要开销；
常见问题：
- Shuffle Write 过大 → 导致 OOM 或 GC 停顿；
- 可通过增加 spark.sql.shuffle.partitions 或启用 AQE 来缓解。

j、`Failure Reason`（仅在 `Failed Stages` 中出现）

含义：Stage 失败的具体原因。
常见错误类型：
- FetchFailedException: 网络连接断开，无法获取 Shuffle 数据（最常见）；
- TaskLost: Task 被杀死（可能因内存不足、GC 时间过长）；
- OutOfMemoryError: JVM 内存溢出；
- Connection closed: 网络异常（如网络抖动、防火墙）；
解决思路：
- 检查网络稳定性；
- 增加 Executor 内存；
- 调整 spark.sql.shuffle.partitions；
- 使用 --conf spark.sql.adaptive.enabled=true 自动优化。

k、`Skipped Stages`

含义：被跳过的 Stage（未执行）。
原因：
- 数据为空；
- 条件不满足（如 if (data.isEmpty()) return;）；
- 作业优化跳过（如缓存命中）；
影响：不会消耗资源，但需确认是否正常。

l、`Failed Stages`

含义：执行失败的 Stage。
后果：
- 整个Job 可能失败；
- Spark 会尝试重试（最多 spark.task.maxFailures 次，默认 4 次）；
应对措施：
- 查看 Failure Reason；
- 分析日志（stderr）；
- 重启或调整资源配置。

2）分析流程

先看整体：看 Completed/Skipped/Failed 数量，快速定位问题；
再看详情：重点关注 Failed Stages 和 Duration 最长的 Stage；
深入分析：
- 点击 +details 查看 Task 级别的执行情况；
- 查看 Shuffle Read/Write 是否过大；
- 观察是否有 FetchFailedException；
结合配置调优：
- 调整 executor-memory、shuffle.partitions；
- 启用 AQE（Adaptive Query Execution）；
- 合理设置 driver-memory 防止 collect 报错。

3）实战技巧

场景	建议操作
`Shuffle Read`/`Write` 很大	增加 `spark.sql.shuffle.partitions`，启用 AQE
Task 失败多	检查日志，看是否 OOM 或 Fetch 失败
Stage 耗时长	查看是否涉及大量 Shuffle 或大数据集
Skipped Stage	确认是否预期行为（如空数据）

3、监控：`Executors`

1）`Summary`

a、摘要统计

Dead 表示该 Executor 已结束，可能是正常完成或异常退出；
如果 Dead 过多，说明任务频繁重启或失败；
动态分配下，Executor 会自动增减，所以 Dead 是正常的。

这是对所有 Executor 的汇总信息，分为三行：

类别	含义
Active(301)	当前正在运行的 Executor 数量（共 301 个）
Dead(501)	已经退出或死亡的 Executor 数量（共 501 个）
Total(802)	总共启动过的 `Executor` 数量 = Active + Dead

b、`RDD Blocks`

常见场景：cache() 后此值会上升。

含义：当前缓存的 RDD 分区数量。
单位：个数
说明：
- 如果值为 0，表示没有数据被缓存；
- 若值较大，说明有数据被 cache() 或 persist()；
- 可用于判断是否启用了缓存策略。

c、`Storage Memory`

Spark 内存分为 Storage（缓存）、Execution（计算中间结果）、Off-Heap 等。

含义：Executor上用于存储缓存数据的内存使用情况。
格式：已用 / 总量（如 0.0 B / 19.1 GB）
用途：
- 查看是否有缓存溢出风险；
- 若接近上限，可能触发 GC 或 OOM；
- 注意：Storage Memory ≠ JVM 堆内存，而是 Spark 管理的内存池。

d、`Disk Used`

含义：Executor 使用的磁盘空间（通常是 Shuffle 文件或临时文件）。
单位：B / MB / GB
注意：
- 大部分情况下应为 0.0 B；
- 若非零，说明有大量数据写入磁盘（如 Shuffle 缓冲区满时）；
- 高于阈值可能导致磁盘 I/O 成瓶颈。

e、`Cores`

含义：该 Executor 拥有的 CPU 核心数。
单位：核数
常见值：4、8、16（取决于配置）
影响：
- 决定并发任务能力；
- executor-cores=4 → 每个 Executor 最多并行跑 4 个 Task。

f、`Active Tasks`

含义：当前正在执行的任务数量。
单位：个数
意义：
- 若长期大于 0，说明还在处理；
- 若某 Executor 一直保持高活跃度，可能是热点问题；
- 结合 Failed Tasks 判断是否卡住。

g、`Failed Tasks`

含义：该 Executor 上失败的任务数量。
单位：个数
重点观察：
- 若某个 Executor 的 Failed Tasks > 0，说明它有问题；
- 可能原因：OOM、GC 时间过长、网络断开；
- 点击 stderr 查看错误日志。

h、`Complete Tasks`

含义：该 Executor 上成功完成的任务数量。
单位：个数
用途：
- 评估工作量分布；
- 若某些 Executor 完成任务远少于其他，说明负载不均。

i、`Total Tasks`

含义：该 Executor 上总共执行过的任务总数。
公式：Complete + Failed
用途：
- 判断整体任务负担；
- 如 Total=28, Complete=28, Failed=0 → 正常；
- 若 Failed 占比高，说明不稳定。

g、`Task Time` (`GC Time`)

建议：启用 G1GC 并监控 GC 日志。

含义：该 Executor 上所有 Task 的总执行时间（含 GC 时间）。
格式：总耗时 (GC 耗时)，如 26 min (36 s)
关键点：
- GC 时间越长，性能越差；
- 若 GC 时间占比超过 10%，需调优内存或开启 G1GC；
- 大型作业中，长时间 GC 是常见瓶颈。

h、`Input`

含义：该 Executor 从外部读取的数据总量（如 HDFS、S3）。
单位：GB / TB
用途：
- 判断数据读取是否均衡；
- 若某 Executor 输入远高于其他，可能是数据倾斜；
- 可结合 Shuffle Read 分析。

i、`Shuffle Read`

含义：该 Executor 从其他节点拉取的 Shuffle 数据量。
单位：GB / MB
重要性：
- 是网络 I/O 的主要来源；
- 若某个 Executor 的 Shuffle Read 极大，可能是 Join 倾斜；
- 常见于 reduceByKey、join 操作。

j、`Shuffle Write`

含义：该 Executor 向其他节点发送的 Shuffle 数据量。
单位：GB / MB
注意：
- 写入越多，磁盘压力越大；
- 若某 Executor 写入远高于平均，可能是数据倾斜源；
- 可通过 spark.sql.adaptive.skewJoin.enabled=true 自动优化。

k、`Blacklisted`

若多个 Executor 被 blacklisted，说明集群不稳定。

含义：该 Executor 是否被标记为“黑名单”（即不再分配新任务）。
值：0 或 1
原因：
- 失败次数过多；
- 网络超时；
- 资源不足；
后果：Spark 不再向其提交新任务，避免重复失败。

2）`Executors` 列表详解

字段	含义
Executor ID	执行器编号（如 `driver`, `1`, `2`…）
Address	执行器所在机器的 IP 和端口（可点击跳转到具体节点）
Status	当前状态： • `Active`：正在运行 • `Dead`：已终止
RDD Blocks	缓存的 RDD 分区数
Storage Memory	使用/总内存
Disk Used	使用的磁盘空间
Cores	CPU 核心数
Active Tasks	正在运行的任务数
Failed Tasks	失败任务数
Complete Tasks	成功任务数
Total Tasks	总任务数
Task Time (GC Time)	总执行时间与 GC 时间
Input	读取的数据量
Shuffle Read	拉取的 Shuffle 数据
Shuffle Write	发送的 Shuffle 数据
Logs	包含 `stdout` 和 `stderr`，点击可查看输出日志

3）分析流程

先看 Summary：快速了解整体情况；
再看 Executor 列表：找出异常节点；
点击 Logs：查看 stderr 获取错误详情；
结合 Stages 页面：确认哪个 Stage 导致了失败；
记录典型模式：比如 “某个 Executor 的 Shuffle Write 特别大”，形成诊断经验。

4）实战技巧

问题类型	观察指标	解决方案
`Executor OOM`	`Storage Memory` 接近上限，`Failed Tasks > 0`	增加 `executor-memory`，减少缓存
`GC` 时间过长	`Task Time (GC Time)` 中 GC 占比高	开启 `G1GC`，调整堆大小
数据倾斜	某个 `Executor` 的 `Shuffle Read` 或 `Write` 明显偏大	启用 `AQE`，广播小表，预聚合
任务失败多	`Failed Tasks` 高，`stderr` 有异常	查看日志，检查网络或代码逻辑
负载不均	某些 `Executor` 的 `Complete Tasks` 很少	检查分区是否均匀，避免热点
`Executor` 死亡多	`Dead` 数量多，`Blacklisted=1`	检查 `YARN` 资源、网络稳定性

ContactAuthor