如何拆分一行文本并将该行附加到Scala / Spark中的每个元素

Question

给出一个像“苹果从树上掉下来”这样的字符串，如何将它分开，使得每个单词都附加了一行文字，这样我得到的字符串RDD看起来像：

"The | The apple fell from a tree"
"apple | The apple fell from a tree"
"fell | The apple fell from a tree"
"from | The apple fell from a tree"
"a | The apple fell from a tree"
"tree | The apple fell from a tree"

这样我就可以跟踪单词的来源。

这是我写的（相关部分）

var inputPath = /path/to/file.txt // Some txt file
var input = sc.textFile(inputPath) // RDD of lines of text
var words = input.flatMap(line => line.split(" ").foreach(word => word.concat(" | " + line)) 

这个代码示例不起作用，因为根据我的理解，你不能在flatMap中多次遍历？我相信我得到一个错误说Found: Unit Required: TraversableOnce[?]我是Spark，Scala和函数式编程的新手。第一次写scala，我不是非常担心性能，或者是最短的代码等等。我只是想要一些工作而不必重新设计我的实现。我以后总能重构。

我知道textFile（）给了我一个带有表示文本每一行的字符串的RDD。 flatMap将这些行拆分为“”，因为它是一个flatMap，我们得到一个数组，而不是一堆数组。如果我错了，或者没有正确说话，请纠正我。

Answer 1

我手边没有火花，所以现在无法确认，但是查看代码和错误信息，很可能只是foreach。

因此，快速修复（可能）将替换最后一行

input.flatMap(line => line.split(" ").map(word => word.concat(" | " + line))

说明：

• line.split给你一个Array[String]，我相信是Traversable[String]的一个例子
• foreach对每个项应用一个函数，但返回一个Unit - 这意味着调用没有返回值（实际上它是Unit类型的单例实例，但是如果它有帮助你可以把它想象成Java语言中的void ）
• map还将函数应用于每个项目，但返回包含更新项目的新Traversable（可能具有不同的具体类型）。
• 最后，flatMap是一种基本上结合map和flatten的方法 - 即它采用一个函数，它接受一个项目并返回一个Traversable[OtherType]，将函数应用于每个项目，而不是通过连接内部遍历来“展平”生成的Traversable[Traversable[OtherType]]。所以你需要给它String => Traversable[String]，但你通过String => Unit

有关更多信息，请参阅Scala Traversable docs。

普通字符串列表上的类似代码：

scala> List(
    "line1 word1 word2", 
    "line2 word3 word4"
)
    .flatMap(line => line.split(" ").map(word => s"$word | $line"))

res5: List[String] = List(
    line1 | line1 word1 word2,
    word1 | line1 word1 word2,
    word2 | line1 word1 word2,
    line2 | line2 word3 word4,
    word3 | line2 word3 word4,
    word4 | line2 word3 word4
)

顺便说一句，Scala鼓励不变性，所以你可能想用val而不是var，除非你真的想重新赋值 - val或多或少类似于final。在您的代码示例中，您可以安全地用vars替换vals。