企业网站改版方案微信营销
Spark计算性能优化实战指南
1 业务场景描述
在某电商平台的实时推荐系统中，需要对历史日志和用户行为数据（规模超过10TB）进行每日批量与小时级别增量计算，为推荐模型提供特征和指标支撑。随着业务量增长，Spark作业执行时间从2小时上升到5小时，资源利用率偏低且容易发生OOM和Task长尾，严重影响上线节奏和用户体验。
2 技术选型过程
- 框架选择：经过对Flink、Presto、Spark三者对比，考虑到Spark生态成熟、社区活跃、与Hadoop生态兼容性好，最终选定Spark 3.x。
- 部署模式：在YARN集群上以动态资源分配（Dynamic Allocation）方式运行，以充分利用集群资源并支持自动伸缩。
- 序列化方式：默认Java序列化效率低、内存占用高，选用Kryo序列化并预注册自定义类，减少GC压力。
- Shuffle服务：启用外部Shuffle服务（spark.shuffle.service.enabled=true），提高任务失败恢复及数据重用能力。
3 实现方案详解
3.1 数据倾斜识别与处理
- 倾斜点识别：通过Spark UI查看Stage中的Shuffle Read Size、Task耗时分布，找到热点Key。
- 解决方案：使用Key Salting方案，即在倾斜Key后追加随机前缀：
// 在Scala中对倾斜列加盐
val saltNum = 10
val salted = rawData.withColumn("salt", (rand()*saltNum).cast("int")).withColumn("joinKey", concat(col("originalKey"), lit("_"), col("salt")))
随后在聚合后再去除前缀还原：
val result = salted.groupBy("joinKey").agg(sum("value").as("sumValue")).withColumn("originalKey", split(col("joinKey"), "_").getItem(0)).groupBy("originalKey").agg(sum("sumValue"))
3.2 广播变量与Join优化
对于小表（<200MB）
val smallDF = spark.read.parquet(smallPath)
val broadcastDF = broadcast(smallDF)
val joined = largeDF.join(broadcastDF, Seq("id"), "left")
- 优势：避免Shuffle，提升Join效率。
- 注意：控制小表大小，避免Broadcast OOM。
3.3 Shuffle分区与并行度调优
spark.default.parallelism=500
spark.sql.shuffle.partitions=500
- 原则：Task数要远小于Executor核心数*2，防止过多小Task，并保持每个Task处理数据量适中（100-300MB）。
3.4 内存管理与GC调优
spark.memory.fraction=0.6
spark.memory.storageFraction=0.3
spark.executor.extraJavaOptions=-XX:+UseG1GC -XX:InitiatingHeapOccupancyPercent=35
- Unified模式：将执行内存与存储内存整合，动态分配避免缓存阻塞计算。
- G1 GC：适合大内存场景，降低Full GC停顿。
3.5 序列化与压缩
spark.serializer=org.apache.spark.serializer.KryoSerializer
spark.kryo.registrationRequired=true
spark.kryo.classesToRegister=com.example.MyClass,org.apache.hadoop.io.Text
spark.io.compression.codec=lz4
- Kryo：比Java序列化速度快、体积小。
- 压缩：使用LZ4兼顾速度和压缩率。
3.6 动态资源分配
spark.dynamicAllocation.enabled=true
spark.dynamicAllocation.minExecutors=10
spark.dynamicAllocation.maxExecutors=200
spark.dynamicAllocation.executorIdleTimeout=60s
- 优势：根据作业阶段自动伸缩Executor数量，提高资源利用率。
4 踩过的坑与解决方案
4.1 广播表过大导致OOM
- 问题：作业中曾将500MB的维度表直接广播，导致Executor OOM。
- 解决：拆分维度表或使用Join Hint，只在关键阶段广播小表。
4.2 Shuffle文件过多
- 问题：默认shuffle分区过多，产生数万个小文件，影响下游读写性能。
- 解决：合理设置spark.sql.shuffle.partitions，并在写入HDFS时合并小文件：
df.coalesce(200).write.parquet(outputPath)
4.3 GC频繁引发长暂停
- 问题：使用Parallel GC时，Full GC停顿超过5s，影响任务执行。
- 解决：切换G1 GC，调整InitiatingHeapOccupancyPercent，并合理分配内存比例。
4.4 数据倾斜重现
- 问题：简单Salting后仍有部分Task耗时过长。
- 解决：结合Map-Side Aggregation与二次Salting，在join前先本地预聚合进一步分散数据。
5 总结与最佳实践
- 合理分区与并行度：根据集群CPU核数和数据量确定shuffle分区；配合coalesce避免小文件。
- 序列化和压缩：优先使用Kryo和LZ4，提高网络和磁盘传输效率。
- 数据倾斜：Salting+预聚合双管齐下；实时监控Spark UI，及时发现热点。
- 内存与GC：Unified模式结合G1 GC，减少Full GC；动态资源分配提高利用率。
- 持续监控与指标采集：使用Spark Listener自定义监控，结合Prometheus+Grafana实时告警。
通过上述实战经验，能够有效提高Spark作业的性能和稳定性，帮助开发者在生产环境中应对大规模数据处理挑战。