如何查询网站的外链寻找客户的渠道和方法

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 数据倾斜识别与处理

1. 倾斜点识别：通过Spark UI查看Stage中的Shuffle Read Size、Task耗时分布，找到热点Key。
2. 解决方案：使用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 总结与最佳实践

1. 合理分区与并行度：根据集群CPU核数和数据量确定shuffle分区；配合coalesce避免小文件。
2. 序列化和压缩：优先使用Kryo和LZ4，提高网络和磁盘传输效率。
3. 数据倾斜：Salting+预聚合双管齐下；实时监控Spark UI，及时发现热点。
4. 内存与GC：Unified模式结合G1 GC，减少Full GC；动态资源分配提高利用率。
5. 持续监控与指标采集：使用Spark Listener自定义监控，结合Prometheus+Grafana实时告警。

通过上述实战经验，能够有效提高Spark作业的性能和稳定性，帮助开发者在生产环境中应对大规模数据处理挑战。