scala – Spark闭包参数绑定

2023年5月8日 263次阅读

我在
Scala中使用Apache Spark.

尝试使用来自第二个RDD的数据操作一个RDD时,我遇到了问题.我试图将第二个RDD作为参数传递给对第一个RDD“映射”的函数,但似乎在该函数上创建的闭包绑定了该值的未初始​​化版本.

以下是一段更简单的代码,显示了我所看到的问题类型. (我第一次遇到麻烦的真实例子是更大,更难以理解).

我真的不明白Spark闭包的参数绑定规则.

我真正想要的是一种基本的方法或模式,用于如何使用另一个RDD(之前在其他地方构建)的内容来操纵一个RDD.

在下面的代码中,调用Test1.process(sc)将失败并在findSquare中使用空指针访问(因为闭包中的第二个arg绑定未初始化)

object Test1 {

  def process(sc: SparkContext) {
    val squaresMap = (1 to 10).map(n => (n, n * n))
    val squaresRDD = sc.parallelize(squaresMap)

    val primes = sc.parallelize(List(2, 3, 5, 7))

    for (p <- primes) {
      println("%d: %d".format(p, findSquare(p, squaresRDD)))
    }
  }

  def findSquare(n: Int, squaresRDD: RDD[(Int, Int)]): Int = {
    squaresRDD.filter(kv => kv._1 == n).first._1
  }
}

最佳答案 您遇到的问题与闭包或RDD无关,与普遍看法相反,
are serializable.

它只是打破了一个基本的Spark规则,它规定你不能从另一个动作或转换*触发一个动作或转换*这个问题的不同变体已被多次询问.

要理解为什么会这样,你必须考虑架构：

> SparkContext在驱动程序上进行管理
>转换中发生的一切都在工人身上执行.每个工作人员只能访问自己的部分数据,不与其他工作人员通信**.

如果要使用多个RDD的内容,则必须使用组合RDD的转换之一,例如join,cartesian,zip或union.

在这里你最有可能(我不确定为什么你传递元组并只使用这个元组的第一个元素)想要使用广播变量：

val squaresMapBD = sc.broadcast(squaresMap)

def findSquare(n: Int): Seq[(Int, Int)] = {
  squaresMapBD.value
    .filter{case (k, v) => k == n}
    .map{case (k, v) => (n, k)}
    .take(1)
}

primes.flatMap(findSquare)

或笛卡尔：

primes
  .cartesian(squaresRDD)
  .filter{case (n, (k, _)) => n == k}.map{case (n, (k, _)) => (n, k)}

将素数转换为虚拟对(Int,null)和连接会更有效：

primes.map((_, null)).join(squaresRDD).map(...)

但根据你的评论,我假设你对有自然连接条件的情况感兴趣.

根据上下文,您还可以考虑使用数据库或文件来存储公共数据.

另一方面,RDD不可迭代,因此您不能简单地使用for循环.为了能够做这样的事情,你必须先收集或转换为LocalIterator.您也可以使用foreach方法.

*准确地说,您无法访问SparkContext.

** Torrent广播和树聚合涉及执行者之间的通信,因此技术上是可行的.

点赞