apache-spark - 在 Spark 中对可迭代值进行排序

Question

假设我有这个输入数据:

["example.com", Date(2000, 1, 1)] : 100,
["example.com", Date(2000, 2, 1)]: 30,
["example.com", Date(2000, 3, 1)]: 5, 
["xyz.com", Date(2000, 1, 1)]: 20,
["xyz.com", Date(2000, 2, 1)]: 10,
["xyz.com", Date(2000, 3, 1)]: 60]

我想按日期(降序)分组，然后按计数排序，给我一个按日期排序的域列表。

我想结束:

Date(2000, 1, 1), [["example.com", 100], ["xyz.com", 20]]
Date(2000, 2, 1), [["example.com", 30], ["xyz.com", 10]]
Date(2000, 3, 1), [["xyz.com", 60], ["example.com", 5]]

这似乎是一个正常的用例，但我从编程指南中看不到这样做的方法。

我可以 map [[domain, date] count] -> [date, [domain, count]]
这会给我 (K, V)对

Date(2000, 1, 1), ["example.com", 100],
Date(2000, 2, 1), ["example.com", 30],
Date(2000, 3, 1), ["example.com", 5], 
Date(2000, 1, 1), ["xyz.com", 20],
Date(2000, 2, 1), ["xyz.com", 10],
Date(2000, 3, 1), ["xyz.com", 60]

然后 groupByKey ，给我 (K, Iterable<V>)对

[Date(2000, 1, 1), [["example.com", 100], ["xyz.com", 20]]
[Date(2000, 2, 1), [["example.com", 30], ["xyz.com", 10]]
[Date(2000, 3, 1), [["example.com", 5], ["xyz.com", 60]]

我怎样才能在键内排序？

请原谅伪代码，我正在使用 Flambo Clojure 包装器，我不想为了问这个问题而用 Java 重写它!

编辑:每个 Iterable(即域列表)可能会太大而无法放入内存。

编辑2:这都是伪代码。我使用月份名称使其可读，但为了清楚起见，我已将其更改为实际日期。

Answer 1

概括地说，我会做以下事情。 (可能不是 100% 正确，因为我没有编译它，但关闭了。)为了简单起见，我假设你从 RDD[((String,String),Int)] 开始。 .

一、groupBy月份类似于:

.groupBy { case ((_, month), _) => month }

并去掉值中的月份:

.mapValues(_.map { case ((domain, _), count) => (domain, count) })

如果需要按月排序，请定义月的排序:

def monthOfYear(month: String): Int = 
  month match {
     case "January" => 1
     case "February" => 2
     ...
  }

并按月对 RDD 进行排序:

.sortBy { case (month, _) => monthOfYear(month) }

并按降序对域进行排序:

.mapValues(_.toSeq.sortBy{ case (domain, count) => count }(Ordering[Int].reverse))

这是直接而有效的，但存在一个月内所有域计数对必须适合内存的问题。

相反，您可以通过按计数降序排序来重新开始:

.sortBy(p => p._2, false)

然后按月分组。我没有测试过这个，我不认为行为是有保证的，但我希望在实践中，即使在分组之后，元素也会按计数顺序遇到。