Spark 闭包参数绑定

Question

我正在使用 Scala 中的 Apache Spark。

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

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

我不太了解 Spark 闭包的参数绑定规则。

我真正在寻找的是一种基本方法或模式，用于如何使用另一个 RDD 的内容（之前在其他地方构建）来操作一个 RDD。

在以下代码中，调用 Test1.process(sc) 将失败并在 findSquare 中访问空指针（因为闭包中绑定的第二个参数未初始化）

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
  }
}

Answer 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)}

将primes 转换为虚拟对(Int, null) 和join 会更有效：

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

但根据您的 cmets，我假设您对存在自然连接条件的场景感兴趣。

根据上下文，您还可以考虑使用数据库或文件来存储常用数据。

附带说明 RDD 不可迭代，因此您不能简单地使用 for 循环。为了能够做这样的事情，你必须先collect 或转换toLocalIterator。你也可以使用foreach方法。

---

* 准确地说，您无法访问SparkContext。

** Torrent 广播和树聚合涉及执行器之间的通信，因此在技术上是可行的。

Answer 2

RDD 不可序列化，因此您不能在 rdd 转换中使用 rdd。 然后我从未见过用 for 语句枚举 rdd，通常我使用作为 rdd api 一部分的 foreach 语句。

为了合并来自两个 rdd 的数据，您可以利用 join、union 或广播（以防您的 rdd 很小）