如何在spark中检索最小值的记录？

Question

假设我有一个像这样的 RDD -> (String, Date, Int)

[("sam", 02-25-2016, 2), ("sam",02-14-2016, 4), ("pam",03-16-2016, 1), ("pam",02-16-2016, 5)]

我想将它转换成这样的列表 ->

[("sam", 02-14-2016, 4), ("pam",02-16-2016, 5)]

其中值是记录，其中日期是每个键的最小值。做这个的最好方式是什么？

Answer 1

我认为既然您将问题标记为与 Spark 相关，您的意思是 RDD 而不是列表。

将记录放入 2 元组，并将键作为第一个元素，这样您就可以使用 reduceByKey 方法，如下所示：

rdd
  .map(t => (t._1, (t._2, t._3))
  .reduceByKey((a, b) => if (a._1 < b._1) a else b)
  .map(t => (t._1, t._2._1, t._2._2))

或者，为了清晰起见，使用模式匹配：（我总是发现元组的 _* 访问器读起来有点混乱）

rdd
  .map {case (name, date, value) => (name, (date, value))}
  .reduceByKey((a, b) => (a, b) match {
     case ((aDate, aVal), (bDate, bVal)) => 
       if (aDate < bDate) a else b
  })
  .map {case (name, (date, value)) => (name, date, value)}

将 替换a._1 < b._1为适合您正在使用的日期类型的任何比较。

请参阅http://spark.apache.org/docs/latest/programming-guide.html#working-with-key-value-pairs有关reduceByKey 的文档，以及您可以在spark 中使用键/值对执行的其他操作

如果您实际上想使用普通的旧 scala 列表来执行此操作，则以下内容可以工作：

list
  .groupBy(_._1)
  .mapValues(l => l.reduce((a, b) => if(a._2 < b._2) a else b))
  .values
  .toList

为了清晰起见，再次使用模式匹配版本：

list
  .groupBy {case (name, date, value) => name}
  .mapValues(l => l.reduce((a, b) => (a,b) match {
    case ((aName, aDate, aValue), (bName, bDate, bValue)) => 
      if(aDate < bDate) a else b
  }))
  .values
  .toList