title: Spark closure argument binding
body: <p>I am working with Apache Spark in Scala.</p>
  
  <p>I have a problem when trying to manipulate one RDD with data from a second RDD. I am trying to pass the 2nd RDD as an argument to a function being 'mapped' against the first RDD, but seemingly the closure created on that function binds an uninitialized version of that value.</p>
  
  <p>Following is a simpler piece of code that shows the type of problem I'm seeing. (My real example where I first had trouble is larger and less understandable).</p>
  
  <p>I don't really understand the argument binding rules for Spark closures.</p>
  
  <p>What I'm really looking for is a basic approach or pattern for how to manipulate one RDD using the content of another (which was previously constructed elsewhere).</p>
  
  <p>In the following code, calling Test1.process(sc) will fail with a null pointer access in findSquare (as the 2nd arg bound in the closure is not initialized)</p>
  
  <pre><code>object Test1 {
  
    def process(sc: SparkContext) {
      val squaresMap = (1 to 10).map(n =&gt; (n, n * n))
      val squaresRDD = sc.parallelize(squaresMap)
  
      val primes = sc.parallelize(List(2, 3, 5, 7))
  
      for (p &lt;- primes) {
        println("%d: %d".format(p, findSquare(p, squaresRDD)))
      }
    }
  
    def findSquare(n: Int, squaresRDD: RDD[(Int, Int)]): Int = {
      squaresRDD.filter(kv =&gt; kv._1 == n).first._1
    }
  }
  </code></pre>