针对你用Spark 2.2.1处理50GB [ID, timestamp, countryId]数据、提取用户国家变更记录速度慢的问题，我从多个核心维度整理了可落地的优化方案：

1. 优先优化数据存储与预处理 #

• 切换列式存储格式：将原始数据转为Parquet或ORC格式（推荐Parquet，Spark 2.2.1对其优化更成熟）。列式存储支持谓词下推、列裁剪，配合snappy压缩（spark.sql.parquet.compression.codec=snappy），能大幅减少IO开销，比CSV/Text格式快3-5倍。
• 提前清理无效数据：在cleanEvents逻辑中，先过滤掉timestamp为空、countryId无效（如空值、非法编码）的记录；同时执行dropDuplicates("ID", "timestamp")去重，避免后续处理重复数据。
• 预分区写入：如果数据会重复使用，写入时按ID分区（df.write.partitionBy("ID").parquet("path")），后续读取时直接按分区加载，避免全局Shuffle。

2. 优化分区与Shuffle配置 #

• 合理设置Shuffle分区数：Spark默认spark.sql.shuffle.partitions=200，对于50GB数据来说可能过少（导致单分区数据过大）或过多（任务调度开销高）。建议设置为500-800（按每100MB一个分区估算），平衡任务并行度与Shuffle开销。
• 前置分区避免重复Shuffle：在执行窗口函数前，先对cleanEvents按ID重分区：

  val partitionedEvents = cleanEvents.repartition($"ID")
  

  这样后续按ID分组的窗口计算就不会触发全局Shuffle，同一用户的数据会集中在同一个Executor节点处理。

3. 核心逻辑（窗口函数）优化 #

提取国家变更记录的核心是用LAG窗口函数对比前后记录，这里要确保逻辑最简、数据量最小：

import org.apache.spark.sql.expressions.Window

// 仅保留必要列，减少窗口计算的数据传输量
val trimmedEvents = cleanEvents.select("ID", "timestamp", "countryId")

// 定义窗口：按ID分区，timestamp升序
val userWindow = Window.partitionBy("ID").orderBy("timestamp")

val result = trimmedEvents
  .withColumn("prev_country", lag("countryId", 1).over(userWindow))
  // 过滤出首次记录（无前置国家）或国家变更的记录
  .filter($"prev_country".isNull || $"countryId" =!= $"prev_country")
  .drop("prev_country")

• 注意：不要在窗口中包含不必要的列，只保留ID、timestamp、countryId即可，减少窗口函数的计算开销。

4. 处理数据倾斜问题 #

如果存在部分ID数据量极大（比如占总数据的10%以上），会导致Shuffle后单个Executor负载过高，拖慢整体速度：

• 大ID加盐拆分：对倾斜的ID添加随机后缀，拆分后并行处理，最后合并结果：

  import org.apache.spark.sql.functions.{rand, concat, lit, floor}
  
  // 识别大ID（假设提前统计出topN大ID）
  val bigIds = Seq("ID1", "ID2", ...)
  val isBigId = $"ID".isin(bigIds:_*)
  
  // 大ID加盐拆分
  val saltedBigEvents = cleanEvents.filter(isBigId)
    .withColumn("salted_id", concat($"ID", lit("_"), floor(rand()*10)))
    .repartition($"salted_id")
  
  // 小ID直接按ID分区
  val smallEvents = cleanEvents.filter(!isBigId).repartition($"ID")
  
  // 分别处理后合并
  val processedBig = processChange(saltedBigEvents).drop("salted_id")
  val processedSmall = processChange(smallEvents)
  val finalResult = processedBig.union(processedSmall)
  

• 单独处理大ID：如果大ID数量极少，可以单独提取这些ID的数据，用本地模式或更高资源的Executor单独计算，再与其他结果合并。

5. 集群与Spark配置调优 #

• Executor资源配置：根据集群节点规格调整，比如：

  spark.executor.memory=16g
  spark.executor.cores=8
  spark.driver.memory=8g
  

  更大的Executor内存能减少GC次数，更多的cores能提升并行处理能力。
• 开启Kryo序列化：Spark默认用Java序列化，速度慢且体积大。设置：

  spark.serializer=org.apache.spark.serializer.KryoSerializer
  spark.kryo.registrationRequired=false
  

  能大幅提升序列化/反序列化速度，减少Shuffle数据量。
• 启用自适应执行（有限支持）：Spark 2.2.1的自适应执行（AQE）处于早期阶段，但可以尝试开启：

  spark.sql.adaptive.enabled=true
  

  它会根据运行时数据自动调整Shuffle分区数，优化执行计划。

6. 避免不必要的算子开销 #

• 尽量使用DataFrame/DataSet API，而非RDD。DataFrame依赖Catalyst优化器能生成更高效的执行计划，比手动编写的RDD逻辑性能高2-3倍。
• 避免在中间步骤中调用collect()、show()等触发Action的操作，除非必要，否则会打断流水线执行，增加额外开销。

内容的提问来源于stack exchange，提问作者Will