Apache Spark 的 reduceByKey() 中的模式匹配 Scala (key, Tuple2) 值答案

【问题标题】：Pattern matching Scala (key, Tuple2) values in reduceByKey() for Apache SparkApache Spark 的 reduceByKey() 中的模式匹配 Scala (key, Tuple2) 值
【发布时间】：2015-07-08 19:00:08
【问题描述】：

我有一个包含 (stockName, stockValue) 元组的 RDD。许多股票是重复的并且具有不同的价值。

例如("ARM", 200.6)("GOOG", 4000.4)("ARM", 3998.23)("ARM", 4002.45)等

这个想法是收集所有类似的股票并计算它们的平均值。

在下面的代码中，map 将每只股票转换为(key, (total, 1))

例如("ARM", (200.6, 1))

reduceByKey 聚合所有同名股票，并独立汇总它们的值和计数。使计算每只股票的平均值变得容易（代码未显示）。

val partial = stocks.map{ case(stock: String, value: Double) => (stock, (value, 1)) } .reduceByKey( (x, y) => (x._1 + y._1, x._2 + y._2) )

在map 中我已经能够使用模式匹配来清楚地表达转换。我希望能够对传递给reduceByKey 的函数参数做同样的事情，以使其更具可读性。

到目前为止，我无法改进 (x, y) => (x._1 + y._1, x._2 + y._2)。

有什么建议吗？

【问题讨论】：

stocks.groupBy(_._1).mapValues(_.foldLeft((0.0, 0))((total, next) => (total._1 + next._2, total._2 + 1))) // Map(ARM -> (8201.279999999999,3), GOOG -> (4000.4,1))

标签： scala apache-spark

【解决方案1】：

您可以嵌套模式以将(x, y) 解构为((x1, x2), (y1, y2))

val partial = stocks.map {
  case (stock, value) => stock -> (value, 1)
}.reduceByKey {
  case ((value1, count1), (value2, count2)) => (value1 + value2, count1 + count2)
}

【讨论】：

谢谢！我曾尝试过，但事实证明我遇到了一个不相关的问题。我实际上是在做一个.reduceByKeyAndWindow（代码是Spark Streaming 应用程序的一部分）并且传递了一个额外的Seconds(20) 参数，代码看起来像：.reduceByKeyAndWindow( (x, y) => (x._1 + y._1, x._2 + y._2), Seconds(20) )。要使用模式匹配版本，我需要插入一些额外的{}，代码现在看起来像：reduceByKeyAndWindow( {case((value1, count1), (value2, count2)) => (value1 + value2, count1 + count2)}, Seconds(20) )。这行得通！我正在慢慢地从 Java 过渡到 Scala！