大数据面试问答-批处理性能优化

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 分区和分桶

分桶的核心价值：通过物理存储的预分区，将相同 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, value
    FROM 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）。