如何将 Spark 上下文从 foreach 传递给函数

Question

我需要将 SparkContext 传递给我的函数，请建议我如何在以下情况下执行此操作。

我有一个序列，每个元素都指向特定的数据源，我们从中获取 RDD 并处理它们。我已经定义了一个函数，它接受 spark 上下文和数据源并执行必要的操作。我目前正在使用 while 循环。但是，我想用 foreach 或 map 来做，这样我就可以暗示并行处理。我需要为函数触发上下文，但我如何从 foreach 传递它？

只是一个示例代码，因为我无法提供实际代码：

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.sql.SQLContext

object RoughWork {
  def main(args: Array[String]) {

    val str = "Hello,hw:How,sr:are,ws:You,re";
    val conf = new SparkConf
    conf.setMaster("local");
    conf.setAppName("app1");
    val sc = new SparkContext(conf);
    val sqlContext = new SQLContext(sc);

    val rdd = sc.parallelize(str.split(":"))
    rdd.map(x => {println("==>"+x);passTest(sc, x)}).collect();

  }

  def passTest(context: SparkContext, input: String) {
    val rdd1 = context.parallelize(input.split(","));
    rdd1.foreach(println)
  }
}

Answer 1

你不能像那样传递 SparkContext。 passTest 将在执行程序上运行，而 SparkContext 在驱动程序上运行。

如果我必须进行这样的双重拆分，一种方法是使用flatMap：

rdd
  .zipWithIndex
  .flatMap(l => {
    val parts = l._1.split(",");
    List.fill(parts.length)(l._2) zip parts})
  .countByKey

可能有更漂亮的方法，但基本上想法是您可以使用zipWithIndex 来跟踪项目来自哪一行，然后使用键值对 RDD 方法处理您的数据。

如果您有多个键，或者通常只有更多结构化数据，您可以考虑将 Spark SQL 与 DataFrames（或最新版本的 DataSets）和 explode 一起使用，而不是 flatMap。