行的 RDD 元素之间的平均值答案

【问题标题】：Average between elements of RDD of rows行的 RDD 元素之间的平均值
【发布时间】：2016-11-14 02:03:47
【问题描述】：

我有一个包含许多行的 RDD（即RDDmacReturns），遵循这种结构：

case class macReturns (macAddress: String, 
                       hourReturns: Long, 
                       threeHoursReturns: Long,
                       sixHoursReturns: Long, 
                       halfDailyReturns: Long, 
                       dailyReturns: Long,
                       threeDailyReturns: Long, 
                       weeklyReturns: Long, 
                       biWeeklyReturns: Long, 
                       threeWeeklyReturns: Long, 
                       monthlyReturns: Long)

因此，例如，该 RDD 的一行将是：

macReturns(a2:b2:c3:d3:f4:c5,3,4,1,0,3,4,3,5,1,7)

macAddress 已经被分组，所以它们都是不同的。

现在，我必须创建一个具有单行的新 RDD，在 RDDmacReturns 上执行转换/操作，它遵循相同的上述结构（案例类 MacReturns）并包含一个固定选择的（假）macAddress 和平均值RDDmacReturns的元素之间计算的每个字段，像这样：

macReturns(00:00:00:00:00:00,
           averageHourReturns,
           averageThreeHoursReturns,
           averageSixHoursReturns,
           averageHalfDailyReturns,
           averageDailyReturns,
           averageThreeDailyReturns,
           averageWeeklyReturns,
           averageBiWeeklyReturns,
           averageThreeWeeklyReturns,
           averageMonthlyReturns)

总而言之，我需要一个应用于 RDDmacReturns 的函数，返回 RDDaverageReturns，它包含一行（上面已经描述过）

感谢您的帮助

【问题讨论】：

你可能想要aggregate和spark.apache.org/docs/latest/api/scala/…
你不能只使用colStats 吗？
@mtoto，我不知道 - 谢谢（不是 OP）。
@mtoto 我也不知道，现在我将尝试操纵我的数据并使用 colStats 计算每列的平均值。谢谢
@mtoto 任务完成，谢谢。只需按照您链接的文档中的示例进行操作就非常容易

标签： scala apache-spark rdd

【解决方案1】：

您可以使用colStats()，它返回MultivariateStatisticalSummary 的一个实例，其中包含按列排列的mean。这是与您的问题类似的可重现示例：

import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.mllib.stat.{MultivariateStatisticalSummary, Statistics}

val rdd = sc.parallelize(Seq(
  ("id1",1,2,3,4),
  ("id2",3,5,1,5),
  ("id3",3,0,9,8),
  ("id4",4,4,1,2)))
// First we convert to RDD of dense vectors 
val rdd_dense = rdd.map(x => Vectors.dense(x._2, x._3, x._4, x._5))
// Attain colStats and grab the mean
val summary: MultivariateStatisticalSummary = Statistics.colStats(rdd_dense)
println(summary.mean) 
[2.75,2.75,3.5000000000000004,4.75]

【讨论】：