如何用 Spark RDD 中的唯一值填充 Scala Seq of Sets？

Question

我正在使用 Spark 和 Scala。我有一个 Array[String] 的 RDD，我将对其进行迭代。 RDD 包含(name, age, work, ...) 等属性的值。我正在使用可变字符串集序列（称为attributes）来收集每个属性的所有唯一值。

把RDD想象成这样：

("name1","21","JobA")
("name2","21","JobB")
("name3","22","JobA")

最后我想要这样的东西：

attributes = (("name1","name2","name3"),("21","22"),("JobA","JobB"))

---

我有以下代码：

val someLength = 10
val attributes = Seq.fill[mutable.Set[String]](someLength)(mutable.Set())
val splitLines = rdd.map(line => line.split("\t"))

lines.foreach(line => {
  for {(value, index) <- line.zipWithIndex} {
    attributes(index).add(value)
    // #1
  }
})

// #2

当我调试并停在标有 #1 的行时，一切都很好，attributes 正确填充了唯一值。

但是在循环之后，在第 #2 行，attributes 再次为空。调查显示，该属性是一系列集合，大小均为 0。

Seq()
Seq()
...

---

我做错了什么？是否有某种我不知道的范围界定？

Answer 1

答案在于 Spark 是一个分布式引擎。我会给你一个关于你所面临的问题的粗略想法。这里每个RDD 中的元素都被分桶到Partitions 中，每个Partition 都可能存在于不同的节点上。

当您编写rdd1.foreach(f) 时，f 被包裹在一个闭包中（它会获取相应对象的副本）。现在，这个闭包被序列化，然后被发送到每个节点，它被应用于Partition 中的每个元素。

在这里，您的f 将在其包装的闭包中获得attributes 的copy，因此当执行f 时，它会与attributes 的副本交互，而不是与您想要的attributes 交互.这会导致您的attributes 被忽略而没有任何更改。

我希望问题现在已经清楚了。

val yourRdd = sc.parallelize(List(
    ("name1","21","JobA"),
    ("name2","21","JobB"),
    ("name3","22","JobA")
))

val yourNeededRdd = yourRdd
  .flatMap({ case (name, age, work) => List(("name", name), ("age", age), ("work", work)) })
  .groupBy({ case (attrName, attrVal) => attrName })
  .map({ case (attrName, group) => (attrName, group.toList.map(_._2).distinct })

// RDD(
//     ("name", List("name1", "name2", "name3")),
//     ("age", List("21", "22")),
//     ("work", List("JobA", "JobB"))
// )

// Or

val distinctNamesRdd = yourRdd.map(_._1).distinct
// RDD("name1", "name2", "name3")

val distinctAgesRdd = yourRdd.map(_._2).distinct
// RDD("21", "22")

val distinctWorksRdd = yourRdd.map(_._3).distinct
// RDD("JobA", "JobB")