排序星火中的可迭代值

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 (即域列表)可能太大，无法在内存中使用。

EDIT2:，这都是psuedocode。我使用月份名称来使其可读性，但为了清晰起见，我已经将其更改为真正的日期。

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)

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