scala - reduceByKey 以 case 类实例为键

Question

我正在使用 Spark 版本 2.4.7-amzn-1 和 Scala 版本 2.11.12(OpenJDK 64 位服务器虚拟机，Java 1.8.0_302)开发 AWS EMR。

我想通过键减少自定义案例类 Item 的数据集，其中键本身是一个自定义案例类。然而，reduceByKey 并没有像我预期的那样工作。

这是两个类:

case class Key(
  name: String,
  color: String
)

case class Item(
  name: String,
  color: String,
  count: Int
) {
    def key: Key = Key(name, color)
}

为了聚合，我在 Item 的伴随对象中定义了一个自定义组合函数，它只是将计数相加:

object Item {
  def combine(i1: Item, i2: Item): Item = i1.copy(count = i1.count + i2.count)
}

这是我的聚合函数:

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

def aggregate(items: Dataset[Item]): Dataset[Item] = items
  .rdd
  .keyBy(_.key)
  .reduceByKey(Item.combine)
  .map(_._2)
  .toDS

现在如果我尝试聚合...

val items: Dataset[Item] = spark.sparkContext.parallelize(
  Seq(
    Item("Square", "green", 8),
    Item("Triangle", "blue", 3),
    Item("Square", "green", 5),
    Item("Triangle", "blue", 7)
  )
).toDS

val aggregated: Dataset[Item] = aggregate(items)

aggregated.show

...输出显示数据集没有减少:

+--------+-----+-----+
|    name|color|count|
+--------+-----+-----+
|  Square|green|    8|
|  Square|green|    5|
|Triangle| blue|    3|
|Triangle| blue|    7|
+--------+-----+-----+

但是，当我更改序列中 4 项的顺序时，我观察到聚合确实有效，因此结果不一致。

如果我将键从案例类实例更改

def key: Key = Key(name, color)

成为一个元组

def key: Tuple2[String, String] = (name, color)

聚合按预期工作，给出以下输出:

+--------+-----+-----+
|    name|color|count|
+--------+-----+-----+
|  Square|green|   13|
|Triangle| blue|   10|
+--------+-----+-----+

那么，reduceByKey 通常不能(可靠地)用于案例类吗？这是预期的行为吗？或者这与案例类与元组无关，而真正的原因隐藏在其他地方？我的 Key 类对我来说似乎很简单，所以我想，这不是散列或比较问题。 (我可能是错的。)

我也看了这个问题reduceByKey using Scala object as key ，但结果证明是拼写错误，chrisbtk 明确指出:“Spark 知道如何比较两个对象，即使它们没有实现 Ordered。”

我总是必须使用元组作为键吗？

Answer 1

尝试直接使用数据集 API:

拥有:

import sparkSession.implicits._
import org.apache.spark.sql.Encoders

implicit val key: Encoder[Key] = Encoders.product[Key]

你可以这样做:

items
  .groupByKey(_.key)
  .reduceGroups(Item.combine)
  .map(_._2)