Spark 性能优化全景：把 40 分钟的 join 调到 5 分钟，我做了这 6 件事

// 默认：HashPartitioner（按 key 的 hashCode % numPartitions）
rdd.partitionBy(new HashPartitioner(100))

// RangePartitioner：按 key 范围分区（适合 sortByKey）
rdd.partitionBy(new RangePartitioner(100, rdd))

// 自定义：业务键的散列
class BusinessPartitioner(numParts: Int) extends Partitioner {
  override def numPartitions: Int = numParts
  override def getPartition(key: Any): Int = {
    val k = key.asInstanceOf[String]
    // 例：业务键 "region#userId" 取 region 部分分区
    val region = k.split("#")(0)
    region.hashCode % numPartitions
  }
}

// 一个有 100 万 key 的 RDD，groupByKey 后只有 1 个 partition 跑
val rdd = ...
rdd.groupByKey().count()  // 慢得离谱

val rdd = sc.parallelize(data, numSlices = 100)  // <-- 100 个 partition
rdd.groupByKey().count()  // 快多了

// 推荐每个分区 10-100 MB
val partitionSize = df.rdd.partitions.length   // <-- 看 RDD 的分区数
// 数据总大小 / 分区数 ≈ 每分区大小

// repartition：总是 shuffle，可以增/减分区
val rdd1 = rdd.repartition(200)        // <-- 强制 shuffle 到 200 个分区

// coalesce：默认不 shuffle，只能减分区
val rdd2 = rdd.coalesce(10)            // <-- 合并到 10 个分区（默认不 shuffle）
val rdd3 = rdd.coalesce(10, shuffle = true)  // <-- 强制 shuffle

val rdd = ... // 100 个分区
val coalesced = rdd.coalesce(10)
// <-- 看似合并到 10 个，但...
coalesced.map(_ * 2).count()  // <-- 结果正确

// groupByKey：把所有 value 拉到一台机器（map 端不合并）
rdd.groupByKey().mapValues(_.sum)  // ⚠️ 大数据 OOM 风险

// reduceByKey：在 map 端先合并，再跨节点 reduce
rdd.mapValues(v => (v, 1))
 .reduceByKey((a, b) => (a._1 + b._1, a._2 + b._2))  // ✅ map 端合并

// aggregateByKey：自定义 zero value + seqOp + combOp
val zero = (0, 0)
rdd.aggregateByKey(zero)(
  (acc, v) => (acc._1 + v, acc._2 + 1),                  // <-- seqOp：map 端合并
  (a, b) => (a._1 + b._1, a._2 + b._2)                    // <-- combOp：跨节点合并
)

// combineByKey：最灵活
rdd.combineByKey(
  v => (v, 1),                                              // <-- createCombiner：单个 value → (sum, count)
  (acc: (Int, Int), v) => (acc._1 + v, acc._2 + 1),        // <-- mergeValue：合并到已有 (sum, count)
  (a: (Int, Int), b: (Int, Int)) => (a._1 + b._1, a._2 + b._2)  // <-- mergeCombiners：跨节点合并
)

// ❌ 错的写法：性能差
rdd.groupByKey().filter { case (_, vals) => vals.size > 100 }.count()
// 整个 groupByKey 的结果都在内存里

// ✅ 改 reduceByKey 后
rdd.mapValues(v => 1).reduceByKey(_ + _).filter { case (_, cnt) => cnt > 100 }.count()
// map 端合并后，shuffle 的数据量大幅减少

// case class：每次 .copy 都是新对象
case class Point(x: Double, y: Double)
val points = sc.parallelize(...).map(p => Point(p.x + 1, p.y + 1))  // <-- 大量小对象

// mutable class：复用对象，但不可变语义丢失
class MutablePoint(var x: Double, var y: Double) {
  def shift(dx: Double, dy: Double): MutablePoint = { x += dx; y += dy; this }
}
val mutablePoints = sc.parallelize(...).map(p => new MutablePoint(p.x, p.y))
//                                <-- for循环复用同一个对象

// Array[Int]：完全避免对象分配
val arrPoints = sc.parallelize(...).mapPartitions { iter =>
  val arr = Array.ofDim[Int](1024)            // <-- 复用同一个数组
  iter.map(p => {
    arr(0) = (p.x + 1).toInt
    arr(1) = (p.y + 1).toInt
    arr
  })
}

// 1000 万条记录 × 每个 32 字节对象 = 320 MB GC 压力
// 用 Array[Int] 后：1000 万条 × 8 字节（基本类型） = 80 MB

// map：每次处理一条
rdd.map(process)         // <-- n 次 process 调用

// mapPartitions：每次处理一个迭代器
rdd.mapPartitions(iter => {
  // 在这里可以初始化昂贵的对象（如 TreeSet、HashMap）
  val treeSet = scala.collection.mutable.TreeSet.empty[Int]
  iter.foreach(x => treeSet.add(x))
  // 一次返回
  Iterator(treeSet.toList)
})                       // <-- 1 次 process 调用（但 partition 内并行）

// 用 TreeSet 算 top-N，内存可控
def topN(partition: Iterator[(String, Int)], n: Int): Iterator[(String, Int)] = {
  val treeSet = scala.collection.mutable.TreeSet.empty[(Int, String)](Ordering.by(-_._1))
  partition.foreach { case (k, v) =>
    treeSet.add((v, k))
    if (treeSet.size > n) treeSet.remove(treeSet.last)    // <-- 保持只留 top-N
  }
  treeSet.toIterator.map { case (v, k) => (k, v) }
}

rdd.mapPartitions(iter => topN(iter, 10))  // <-- 每个 partition 算自己的 top-10

// BroadcastHashJoin：小表广播
val result = bigDf.join(smallDF.broadcast(), "user_id")
//                <-- 强制广播

// SortMergeJoin：默认的大表 join
val result = bigDf.join(otherBigDf, "user_id")
//                <-- 默认 SortMergeJoin（全 shuffle）

// Bucket：重复 join 用 bucketBy
bigDf.write.bucketBy(10, "user_id").saveAsTable("bucketed_users")
otherDf.write.bucketBy(10, "user_id").saveAsTable("bucketed_orders")
spark.read.table("bucketed_users").join(spark.read.table("bucketed_orders"), "user_id")
//                                <-- 不再 shuffle！

// 默认阈值是 10MB（spark.sql.autoBroadcastJoinThreshold）
// 调小：更多 join 走 broadcast，但 driver 内存压力大
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "50mb")

// ❌ 倾斜的 join
val result = skewedDf.join(normalDf, "key")
//                                              <-- skewedDf 里 90% 的 key 都是同一个

// 给倾斜 key 加几个虚拟 key
val skewedKey = "popular_key"
val skewedWithSalt = skewedDf.withColumn("salted_key",
  when($"key" === skewedKey, concat($"key", lit("_"), (rand() * 10).cast("int")))
 .otherwise($"key")
)

// 同样的虚拟化处理另一边
val normalWithSalt = normalDf.withColumn("salted_key",
  when($"key" === skewedKey, concat($"key", lit("_"), (rand() * 10).cast("int")))
   .otherwise($"key")
)

val result = skewedWithSalt.join(normalWithSalt, "salted_key")

// 给倾斜 key 加 0-99 的随机后缀
val salted = skewedDf.withColumn("salt", (rand() * 100).cast("int"))
val exploded = salted.withColumn("salt", explode(array($"salt".cast("int"))))

// 对另一边也加 0-99
val result = exploded.join(otherDf, "salt")

// spark.conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
// spark.conf.set("spark.kryo.registerKryoClasses", Array(classOf[MyCaseClass]))
spark.conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

// ❌ 错误：driver 端的对象传到 executor
class NotSerializableConfig(val thresholds: Map[String, Int])

rdd.foreachPartition { partition =>
  val config = new NotSerializableConfig(...)  // ⚠️ 闭包捕获，序列化失败
  ...
}

// ✅ 修法 1：继承 Serializable
class SerializableConfig(val thresholds: Map[String, Int]) extends Serializable
//<-- 让 Spark 能序列化它

// ✅ 修法 2：用局部变量
rdd.foreachPartition { partition =>
  val config = SerializableConfig(thresholds)  // <-- 在闭包内创建，不捕获外层引用
  ...
}

// ✅ 修法 3：用 broadcast
val configBroadcast = spark.sparkContext.broadcast(thresholds)
rdd.foreachPartition { partition =>
  val config = configBroadcast.value
  ...
}

// ❌ 错误代码
val myClass = new MyService("http://example.com")
rdd.foreachPartition { partition =>
  myClass.call(...) // ⚠️ myClass 不可序列化
}

rdd.foreachPartition { partition =>
  val myClass = new MyService("http://example.com")  // <-- 在闭包内
  myClass.call(...)
}

df.explain()                  // <-- 只看物理计划
df.explain(true)              // <-- 物理 + 逻辑
df.explain(mode = "extended") // <-- 物理 + 逻辑 + parsed + analyzed

// 关键节点
// 1. Exchange：shuffle 的代名词
// 2. SortMergeJoin：默认的大表 join
// 3. BroadcastHashJoin：小表进广播的快速 join
// 4. Filter + PushedFilters：谓词下推

spark.read.parquet("path").filter($"year" === 2022).explain(true)
// == Physical Plan ==
// *(1) Filter (isnotnull(year#5) AND (year#5 = 2022))
// +- *(1) ColumnarToRow
//    +- FileScan parquet [year#5] PushedFilters: [IsNotNull(year), EqualTo(year, 2022)]

spark.conf.set("spark.sql.adaptive.enabled", "true")
// <-- 开启自适应执行

// AQE 会自动：
// - 合并小 partition
// - 自动 broadcast 小表
// - 动态调整 join 策略

import org.apache.spark.storage.StorageLevel

df.persist(StorageLevel.MEMORY_ONLY_SER)  // <-- 序列化进内存
df.cache() // <-- 等价于 MEMORY_ONLY

// 用完释放
df.unpersist()

df.cache()  // <-- 占 executor 内存
// 不释放 = 内存一直占用