Spark reduceByKey以case class实例为key聚合异常问题咨询

reduceByKey完全支持case class作为key，你的问题不是Spark的功能限制，而是自定义Key类的hashCode和equals逻辑不符合Spark的shuffle要求，导致出现不稳定的聚合结果：
Spark的reduceByKey依赖两个核心判断逻辑完成聚合：

• 基于key的hashCode计算目标分区，相同key必须分到同一个分区
• 同一分区内基于key的equals判断是否为同一个key，执行合并
  你遇到的偶发成功的现象，本质是相同字段的Key实例偶尔被分到同一个分区、且同分区内判断为相等时才会聚合，大多数时候因为hashCode/equals判断不一致，导致相同逻辑的key被判定为不同，无法合并。

导致问题的常见场景 #

你定义的Key是普通case class，理论上默认会生成基于所有字段的hashCode和equals，出现不符合预期的情况大概率是以下两种原因：

1. Key类定义在非静态上下文
   如果Key被定义在普通类内部、方法内部、或者Spark Shell的会话类内部，case class会隐式持有外部类的实例引用，默认生成的hashCode和equals会把外部类实例纳入计算。即使两个Key的name和color完全相同，只要对应的外部类实例不同，就会被判定为不同的key。
   而Tuple2是Scala标准库的顶级类，不存在外部引用，hashCode和equals完全基于内部元素计算，因此不会出现该问题。
2. 类加载器不一致
   在AWS EMR的分布式环境下，如果Key类没有被正确分发到所有worker节点，或者不同节点加载的Key类属于不同的类加载器，JVM会直接判定两个类实例不相等，即使字段完全一致。

解决方案 #

你不需要强制使用Tuple作为key，以下任意一种方案都可以解决问题：

方案1：调整Key类的定义位置 #

将Key定义为顶级类，或者放在静态object内部，避免隐式持有外部类引用：

// 放在静态object内部，或者直接作为顶级类定义
object Models {
  case class Key(name: String, color: String)
  case class Item(name: String, color: String, count: Int) {
    def key: Key = Key(name, color)
  }
}

方案2：显式重写Key的hashCode和equals #

强制两个方法的计算逻辑仅依赖name和color字段，屏蔽外部引用的影响：

case class Key(name: String, color: String) {
  override def hashCode(): Int = (name, color).hashCode()
  override def equals(obj: Any): Boolean = obj match {
    case other: Key => other.name == name && other.color == color
    case _ => false
  }
}

方案3：直接使用Dataset API避免转RDD #

你不需要将Dataset转为RDD做聚合，直接使用Dataset的groupByKey可以规避RDD层面的key判断问题：

import org.apache.spark.sql.Dataset
import spark.implicits._

def aggregate(items: Dataset[Item]): Dataset[Item] = items
  .groupByKey(i => (i.name, i.color))
  .reduceGroups(Item.combine)
  .map(_._2)

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