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

Question

假设我有一个RDD，格式如下：(字符串, 日期, 整数)。

[("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)]

如何才能找到每个键的最小日期对应的记录呢？

有什么最好的方法吗？

- lapolonio

---

我在考虑使用groupbykey然后再使用takeordered，但我不确定最好的方法是什么，因为groupbykey很慢。 - lapolonio

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

- Angelo Genovese