skillhub-110-spark-optimization
wehub-skill-sync
96 行
3.1 KiB
Markdown
96 行
3.1 KiB
Markdown
---
|
|
name: spark-optimization
|
|
description: 优化 Apache Spark 作业，涵盖分区、缓存、shuffle 优化与内存调优。适用于提升 Spark 性能、调试慢作业或扩缩数据处理管道时使用。
|
|
---
|
|
|
|
# Apache Spark 优化
|
|
|
|
优化 Apache Spark 作业的生产级模式，包括分区策略、内存管理、shuffle 优化与性能调优。
|
|
|
|
## 何时使用此技能
|
|
|
|
- 优化运行缓慢的 Spark 作业
|
|
- 调优内存与执行器配置
|
|
- 实现高效的分区策略
|
|
- 调试 Spark 性能问题
|
|
- 为大数据集扩缩 Spark 管道
|
|
- 减少 shuffle 与数据倾斜
|
|
|
|
## 核心概念
|
|
|
|
### 1. Spark 执行模型
|
|
|
|
```
|
|
Driver 程序
|
|
↓
|
|
作业（由 action 触发）
|
|
↓
|
|
阶段（以 shuffle 分隔）
|
|
↓
|
|
任务（每个分区一个）
|
|
```
|
|
|
|
### 2. 关键性能因素
|
|
|
|
| 因素 | 影响 | 解决方案 |
|
|
| ------------------- | ----------------------- | ------------------------------ |
|
|
| **Shuffle** | 网络 I/O、磁盘 I/O | 减少宽依赖转换 |
|
|
| **数据倾斜** | 任务执行时间不均 | 加盐、广播连接 |
|
|
| **序列化** | CPU 开销 | 使用 Kryo、列式格式 |
|
|
| **内存** | GC 压力、溢出到磁盘 | 调优执行器内存 |
|
|
| **分区数** | 并行度 | 设置合适的分区大小 |
|
|
|
|
## 快速开始
|
|
|
|
```python
|
|
from pyspark.sql import SparkSession
|
|
from pyspark.sql import functions as F
|
|
|
|
# 创建优化后的 Spark 会话
|
|
spark = (SparkSession.builder
|
|
.appName("OptimizedJob")
|
|
.config("spark.sql.adaptive.enabled", "true")
|
|
.config("spark.sql.adaptive.coalescePartitions.enabled", "true")
|
|
.config("spark.sql.adaptive.skewJoin.enabled", "true")
|
|
.config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
|
|
.config("spark.sql.shuffle.partitions", "200")
|
|
.getOrCreate())
|
|
|
|
# 使用优化设置读取数据
|
|
df = (spark.read
|
|
.format("parquet")
|
|
.option("mergeSchema", "false")
|
|
.load("s3://bucket/data/"))
|
|
|
|
# 高效转换
|
|
result = (df
|
|
.filter(F.col("date") >= "2024-01-01")
|
|
.select("id", "amount", "category")
|
|
.groupBy("category")
|
|
.agg(F.sum("amount").alias("total")))
|
|
|
|
result.write.mode("overwrite").parquet("s3://bucket/output/")
|
|
```
|
|
|
|
## 详细模式与示例
|
|
|
|
详细模式文档位于 `references/details.md`。当上述导航层级不够用时，请阅读该文件。
|
|
|
|
## 最佳实践
|
|
|
|
### 应该做的
|
|
|
|
- **启用 AQE** — 自适应查询执行能处理许多问题
|
|
- **使用 Parquet/Delta** — 列式格式并启用压缩
|
|
- **广播小表** — 对小表连接避免 shuffle
|
|
- **监控 Spark UI** — 检查数据倾斜、溢出、GC 情况
|
|
- **设置合适的分区大小** — 每个分区 128MB — 256MB
|
|
|
|
### 不应做的
|
|
|
|
- **不要收集大数据** — 保持数据分布式
|
|
- **不要不必要地使用 UDF** — 优先使用内置函数
|
|
- **不要过度缓存** — 内存是有限的
|
|
- **不要忽视数据倾斜** — 它会主导作业运行时间
|
|
- **不要用 `.count()` 判断是否存在** — 应使用 `.take(1)` 或 `.isEmpty()`
|