scala - 如何在 Spark Scala 中使用 mapPartitions？

Question

我有 DocsRDD : RDD[String, String]

val DocsRDD = sc.wholeTextFiles("myDirectory/*" , 2)

DocsRDD:

Doc1.txt , bla bla bla .....\n bla bla bla \n bla ... bla
Doc2.txt , bla bla bla .....bla \n bla bla \n bla ... bla
Doc3.txt , bla bla bla .....\n bla bla bla \n bla ... bla
Doc4.txt , bla bla \n  .....\n bla bla bla bla \n ... bla

有没有一种高效、优雅的方法来使用 mapPartitions 从这些中提取 n-gram？到目前为止，我已经尝试了所有内容，我已经阅读了至少 5 遍我能找到的关于 mapPartitions 的所有内容，但我仍然无法理解如何使用它!操作起来似乎太难了。总之我想要:

val NGramsRDD = DocsRDD.map(x => (x._1 , x._2.sliding(n) ) )

但使用 mapPartitions 可以有效。我对 mapPartitions 的基本误解是:

OneDocRDD : RDD[字符串]

 val OneDocRDD = sc.textFile("myDoc1.txt" , 2)
                   .mapPartitions(s1 : Iterator[String] => s2 : Iterator[String])

我无法理解这一点!从什么时候 s1 是 Iterator[String]？ s1 是 sc.textfile 之后的字符串。

好吧，我的第二个问题是:在这种情况下，mapPartitions 会改善我对抗 map 的能力吗？

最后但并非最不重要的:f() 可以是:

     f(Iterator[String]) : Iterator[Something else?]

Answer 1

我不确定 .mapPartitions 是否会有所帮助(至少，没有给出示例)，但使用 .mapPartitions 看起来像:

val OneDocRDD = sc.textFile("myDoc1.txt", 2)
  .mapPartitions(iter => {
    // here you can initialize objects that you would need 
    // that you want to create once by worker and not for each x in the map. 
    iter.map(x => (x._1 , x._2.sliding(n)))
  })

通常您想使用 .mapPartitions 来创建/初始化您不想要的对象(例如:太大)或无法序列化到工作节点。如果没有 .mapPartitions，您将需要在 .map 中创建它们，但这不会有效，因为将为每个 x 创建对象。