# 编写优化的SQL
— 优化1：尽早过滤
— 不推荐
SELECT a.*, b.*
FROM a JOIN b ON a.id = b.id
WHERE a.date = ‘2024-04-08’;
— 推荐
SELECT a.*, b.*
FROM (SELECT * FROM a WHERE date = ‘2024-04-08’) a
JOIN (SELECT * FROM b WHERE date = ‘2024-04-08’) b
ON a.id = b.id;

— 优化2：使用广播join
— 小表自动广播（默认10MB）
SELECT /*+ BROADCAST(b) */ a.*, b.*
FROM a JOIN b ON a.id = b.id;

— 优化3：合理设置分区
SET spark.sql.shuffle.partitions=200;

— 优化4：使用开窗函数优化
— 不推荐
SELECT * FROM (
SELECT *, ROW_NUMBER() OVER (PARTITION BY id ORDER BY ts DESC) rn
FROM table
) t WHERE rn = 1;
— 推荐：如果数据量大，考虑其他方式

— 优化5：避免笛卡尔积
— 不推荐
SELECT * FROM a, b;
— 推荐：一定要加join条件

# 解决方案1：过滤倾斜key
val skewedKeys = Array(“key1”, “key2”, “key3”)
val rddWithoutSkew = rdd.filter(x => !skewedKeys.contains(x._1))
val rddSkew = rdd.filter(x => skewedKeys.contains(x._1))
// 分别处理再合并

# 解决方案2：加盐处理
// 给倾斜的key加随机前缀
val rddWithSalt = rdd.map(x => {
if (isSkewedKey(x._1)) {
(x._1 + “_” + Random.nextInt(100), x._2)
} else {
x
}
})
// 聚合后去掉前缀
val result = rddWithSalt.reduceByKey(_ + _)
.map(x => (x._1.split(“_”)(0), x._2))
.reduceByKey(_ + _)

# 解决方案3：广播小表
// 如果是join导致的倾斜
val smallTableDF = spark.read.table(“small_table”)
val broadcastSmallTable = broadcast(smallTableDF)
val result = largeTableDF.join(broadcastSmallTable, “id”)

# 解决方案4：调整并行度
SET spark.sql.shuffle.partitions=500;

# 解决方案5：使用AQE自动处理
— 启用AQE倾斜join优化
SET spark.sql.adaptive.skewJoin.enabled=true;
SET spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes=256MB;

# Spark UI页面
Jobs页：
– 查看所有Job
– 查看Job状态
– 点击Job查看详情

Stages页：
– 查看所有Stage
– 查看Stage状态
– 点击Stage查看Task详情

Storage页：
– 查看缓存的RDD
– 查看内存使用
– 查看缓存大小

Environment页：
– 查看配置参数
– 查看类路径
– 查看系统属性

Executors页：
– 查看Executor列表
– 查看内存使用
– 查看GC时间
– 查看Task统计

SQL页：
– 查看SQL执行
– 查看执行计划
– 查看数据扫描量

# 分析方法
1. 看哪个Job慢
2. 看Job里哪个Stage慢
3. 看Stage里哪个Task慢
4. 看Shuffle数据量
5. 看GC时间
6. 看Spill情况