网站建设成都2023国际别墅设计展
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）。