在 Scalding 中生成 List[String] 的差异答案

【问题标题】：Generate diff of List[String] in Scalding在 Scalding 中生成 List[String] 的差异
【发布时间】：2016-03-29 22:32:04
【问题描述】：

我在我的 Scalding 工作中有一个 records:TypedType[(String, util.List[String])]，其中第一个值是一个 id，第二个是一个东西列表。想象一下：

("1", ["a","b","c"])
("1", ["a","b","c"])
("1", ["a","b","c"])
("2", ["a","b"])
("2", ["a","b","c"])
("3", ["a","b","c"])

records.groupBy(_._1) 之后，我只想输出给定 id 彼此不同的记录。对于上面的输入，输出应该是：

("2", ["a","b"])
("2", ["a","b","c"])

我是 Scalding 的新手。实现这一目标的优雅方法是什么？

【问题讨论】：

标签： java scala scalding

【解决方案1】：

我不知道 Scalding 方面是否对您很重要（您的收藏是否异常庞大？）但在普通的 Scala 中我会这样做：

// Given:
val records = Seq( "1" -> List("a", "b", "c"), "1" -> List("a", "b", "c"), "1" -> List("a", "b", "c"), "2" -> List("a", "b"), "2" -> List("a", "b", "c"), "3" -> List("a", "b", "c"), "3" -> List("d")

val distinctValues = records.groupBy(_._1).map { case (k, v) => k -> v.toSet }
// => Map(2 -> Set((2,List(a, b)), (2,List(a, b, c))), 1 -> Set((1,List(a, b, c))), 3 -> Set((3,List(a, b, c)), (3,List(d))))

val havingMultipleDistinct = distinctValues.map { case (k, v) => v.size > 1 }
// => Map(2 -> Set((2,List(a, b)), (2,List(a, b, c))), 3 -> Set((3,List(a, b, c)), (3,List(d))))

val asRecords = havingMultipleDistinct.values.flatten
// => List((2,List(a, b)), (2,List(a, b, c)), (3,List(a, b, c)), (3,List(d)))

【讨论】：

是的，它必须在集群上运行。烫伤是根本

【解决方案2】：

如果每个键的值的大小足够小以适合内存，那么应该这样做：

records
  .group
  .toSet
  .filter(_.size > 1)
  .flatten

如果太大，可以自己加入管道：

val grouped = records.group
grouped
 .join(grouped)
 .collect { case(k, (a, b)) if a != b => k -> a }

【讨论】：