title: Spark UDF exception when accessing broadcast variable
body: <p>I'm having difficulty accessing a <code>scala.collection.immutable.Map</code> from inside a spark UDF. </p>
  
  <p>I'm broadcasting the map</p>
  
  <pre><code>val browserLangMap = sc.broadcast (Source.fromFile(browserLangFilePath).getLines.map(_.split(,)).map(e =&gt; (e(0).toInt,e(1))).toMap)
  </code></pre>
  
  <p>creating UDF that access the map</p>
  
  <pre><code>def addBrowserCode = udf((browserLang:Int) =&gt; if(browserLangMap.value.contains(browserLang)) browserLangMap.value(browserLang) else "")`
  </code></pre>
  
  <p>using the UDF to add new column</p>
  
  <pre><code>val joinedDF = rawDF.join(broadcast(geoDF).as("GEO"), $"start_ip" === $"GEO.start_ip_num", "left_outer")
                          .withColumn("browser_code", addBrowserCode($"browser_language"))
                          .selectExpr(getSelectQuery:_*)
  </code></pre>
  
  <p>full stack trace --> <a href="https://www.dropbox.com/s/p1d5322fo9cxro6/stack_trace.txt?dl=0" rel="nofollow noreferrer">https://www.dropbox.com/s/p1d5322fo9cxro6/stack_trace.txt?dl=0</a></p>
  
  <pre><code>org.apache.spark.SparkException: Task not serializable
          at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:304)
          at org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:294)
          at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:122)
          at org.apache.spark.SparkContext.clean(SparkContext.scala:2055)
          at org.apache.spark.SparkContext.runJob(SparkContext.scala:1857)
          at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
  Caused by: java.io.NotSerializableException: $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$MetaDataSchema$
  Serialization stack:
          - object not serializable (class: $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$MetaDataSchema$, value: $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$MetaDataSchema$@30b4ba52)
          - field (class: $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC, name: MetaDataSchema$module, type: class $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$MetaDataSchema$)
          - object (class org.apache.spark.sql.catalyst.expressions.ScalaUDF, UDF(browser_language#235))
          - field (class: org.apache.spark.sql.catalyst.expressions.If, name: falseValue, type: class org.apache.spark.sql.catalyst.expressions.Expression)
          - object (class org.apache.spark.sql.catalyst.expressions.If, if (isnull(browser_language#235)) null else UDF(browser_language#235))
          - field (class: org.apache.spark.sql.catalyst.expressions.Alias, name: child, type: class org.apache.spark.sql.catalyst.expressions.Expression)
          - object (class org.apache.spark.sql.catalyst.expressions.Alias, if (isnull(browser_language#235)) null else UDF(browser_language#235) AS browser_language#507)
          - object (class org.apache.spark.OneToOneDependency, org.apache.spark.OneToOneDependency@5ae38c4e)
          - writeObject data (class: scala.collection.immutable.$colon$colon)
          - object (class scala.collection.immutable.$colon$colon, List(org.apache.spark.OneToOneDependency@5ae38c4e))
          - field (class: org.apache.spark.rdd.RDD, name: org$apache$spark$rdd$RDD$$dependencies_, type: interface scala.collection.Seq)
          at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:301)
          ... 80 more
  </code></pre>
  
  <p>I know its the access to broadcast Map that is causing this. When I remove reference to that in the UDF there are no exception.</p>
  
  <pre><code>def addBrowserCode = udf((browserLang:Int) =&gt; browserLang.toString())  //Test UDF without accessing broadcast Map and it works
  </code></pre>
  
  <p>Spark version 1.6</p>