1. 数据存储角度
1.1 存储优化
列式存储格式:使用Parquet/ORC代替CSV/JSON,减少I/O并提升压缩率。
df.write.parquet("hdfs://path/output.parquet")
列式存储减少I/O的核心机制:
列裁剪(Column Pruning)
原理:查询时只读取需要的列,跳过无关列。
示例:
若执行 SELECT AVG(Age) FROM users
,只需读取Age列的数据块,而无需加载Name、City等列。
节省效果:假设表有100列,仅读取1列时,I/O量减少99%。
高效压缩(Compression)
数据局部性:同一列的数据类型和值域相似,压缩效率更高。
数值型数据(如Age):可使用Delta Encoding(存储差值)或Run-Length Encoding(连续重复值压缩)。
字符串数据(如City):可使用字典编码(如NY→1, SF→2)。
1.2 小文件合并
Spark
任务使用coalesce
或repartition
合并任务输出的小文件,
Hive
需手动执行ALTER TABLE COMPACT
1.3 分区和分桶
特性 | 分区(Partitioning) | 分桶(Bucketing) |
---|---|---|
目的 | 减少扫描范围(按目录过滤) | 优化JOIN、采样、数据局部性 |
实现方式 | 按字段值划分目录(如/date=20230101/) | 按字段哈希值分到固定数量的文件(桶) |
语法 | PARTITIONED BY (date STRING) | CLUSTERED BY (user_id) INTO 10 BUCKETS |
适用场景 | 高基数字段(如日期、地域) | 低基数字段(如用户ID、分类ID) |
分桶的核心价值:通过物理存储的预分区,将相同 Key 的数据聚集到同一位置,使得 Hive 可以在 Map 阶段直接完成 JOIN,跳过 Shuffle 和 Reduce。
本质区别:
分区(Partitioning)是 粗粒度过滤(按目录剪枝),减少数据扫描范围。
分桶(Bucketing)是 细粒度分布(按哈希分片),优化数据计算效率。
适用场景:高频大表 JOIN、数据倾斜缓解、高效采样。
2. 计算角度
2.1 join优化
使用Broadcast Join
小表广播与分桶优化Join
,具体可看大数据面试问答-Hadoop/Hive/HDFS/Yarn中的2.3章节
2.2 数据倾斜
2.2.1 加盐处理
对倾斜的Key添加随机前缀,分散数据
-- 原始SQL
SELECT user_id, COUNT(*) FROM logs GROUP BY user_id;-- 加盐优化后
SELECT user_id, SUM(cnt) FROM (SELECT CONCAT(user_id, '_', FLOOR(RAND()*10)) AS salted_key, COUNT(*) AS cnt FROM logs GROUP BY salted_key
) GROUP BY user_id;
使用两阶段聚合(加盐后局部聚合,再去盐全局聚合)
# 原始代码(存在倾斜)
rdd.groupByKey().mapValues(sum)# 优化后(两阶段聚合)
# 1. 加盐并局部聚合
salted_rdd = rdd.map(lambda x: (f"{x[0]}_{random.randint(0,9)}", x[1]))
partial_sum = salted_rdd.reduceByKey(lambda a, b: a + b)
# 2. 去盐并全局聚合
unsalted_rdd = partial_sum.map(lambda x: (x[0].split("_")[0], x[1]))
final_sum = unsalted_rdd.reduceByKey(lambda a, b: a + b)
2.2.2 增加分区
当某些 Key 的数据量极大时,默认的分区数(如 spark.sql.shuffle.partitions=200
)可能导致这些 Key 的数据集中在少数分区中,造成长尾任务。
增加分区数(如设置为 1000)可以将原本集中在少量分区的数据分散到更多分区,降低单个分区的数据量,从而缓解倾斜。
例如:假设一个 Key 的数据量占整体的 50%,当分区数从 200 增加到 1000 时,该 Key 的数据会被 Hash 分配到更多分区(但实际效果受 Hash 算法影响,可能仍不均匀)。
适用场景:
数据倾斜由多个不同的 Key 导致(如多个热 Key),而非单个超大 Key。
分区数不足导致部分分区负载过高(如默认分区数远小于实际 Key 的基数)。
局限性:
对单个超大 Key(如某个 Key 占 90% 数据量)无效,因为该 Key 的所有数据仍会被 Hash 到同一个分区。
此时必须使用 加盐(如为 Key 附加随机前缀,强制分散数据)。
示例:Spark 中的分区优化
# 增加 Shuffle 分区数(全局配置)
spark.conf.set("spark.sql.shuffle.partitions", "1000")# 或在特定操作中重新分区
df.repartition(1000, "key").groupBy("key").count()
示例:Hive 中的优化
-- 增加 Reduce 任务数
SET mapred.reduce.tasks = 1000;-- 两阶段聚合(自动处理倾斜)
SET hive.groupby.skewindata = true;-- 手动加盐处理单个超大 Key
SELECT key_salted, SUM(value)
FROM (SELECT CONCAT(key, '_', CAST(RAND() * 10 AS INT)) AS key_salted, valueFROM input_table
) tmp
GROUP BY key_salted;
2.2.3 过滤无效数据
提前剔除无意义的空值或默认值,减少无效计算
2.2.4 单独处理
拆分倾斜Key单独处理
2.3 算子优化
减少Shuffle:
避免不必要的groupByKey
,优先用reduceByKey
。
3. 资源角度
Executor配置
# 单个Executor资源
--executor-memory 16G # 内存(留20%给堆外内存)
--executor-cores 4 # CPU核心
--num-executors 20 # Executor数量# 动态资源分配(应对数据波动)
spark.dynamicAllocation.enabled=true
内存管理:
调整内存分配比例:spark.memory.fraction=0.6
(默认0.6,Execution和Storage共享)。
堆外内存优化:spark.executor.memoryOverhead=2G
(防止OOM)。