title: udf in Spark SQL
body: <p>I have two dataframes: dataDf and regexDf.  dataDf has large number of records and regexDf has two columns of regex.  My problem is, i need to filter dataDf based on two of its column matching regular expression in two columns in regexDef.  I came up with this</p>
  
  <pre><code>dataDf.registerTempTable("dataTable")
  sqlContext.udf.register("matchExpressionCombination", matchExpressionCombination _)
  
  val matchingResults = sqlContext.sql("SELECT * FROM dataTable WHERE matchExpressionCombination(col1, col2)")
  def matchExpressionCombination(col1Text: String, col2Text: String): Boolean = {
    val regexDf = getRegexDf()
    var isMatch = false
    for(row &lt;- regexDf.collect) {
      if(col1Text.matches(row(0).toString) &amp;&amp; col2Text.matches(row(1).toString)) {
        isMatch = true
      }
    }
    isMatch
  }
  </code></pre>
  
  <p>when i say</p>
  
  <pre><code>matchingResults.count().println
  </code></pre>
  
  <p>​I get the following error:-</p>
  
  <pre><code>Exception in thread "main" org.apache.spark.sql.catalyst.errors.package$TreeNodeException: execute, tree:
  TungstenAggregate(key=[], functions=[(count(1),mode=Final,isDistinct=false)], output=[count#118L])
   TungstenExchange SinglePartition
    TungstenAggregate(key=[], functions=[(count(1),mode=Partial,isDistinct=false)], output=[currentCount#121L])
     Union
      TungstenProject
       Filter UDF(Col1Text#97,Col2Text#109)
        Scan CsvRelation(playWithSpark/data/dataDf1.csv,true,,,",null,#,PERMISSIVE,COMMONS,false,false,null,UTF-8,false)[Col1Text#97,BaselineStacktrace#98,BaselineTime#99,ClassId#100,ClassName#101,Id#102,IsDataSilo#103,MethodName#104,Namespace#105,Organization#106,PackageName#107,TestResultKey#108,Col2Text#109,UpgradedStacktrace#110,UpgradedTime#111]
      TungstenProject
       Filter UDF(Col1Text#2,Col2Text#14)
        Scan CsvRelation(playWithSpark/data/dataDf2.csv,true,,,",null,#,PERMISSIVE,COMMONS,false,false,null,UTF-8,false)[Col1Text#2,BaselineStacktrace#3,BaselineTime#4,ClassId#5,ClassName#6,Id#7,IsDataSilo#8,MethodName#9,Namespace#10,Organization#11,PackageName#12,TestResultKey#13,Col2Text#14,UpgradedStacktrace#15,UpgradedTime#16]
  
  
      at org.apache.spark.sql.catalyst.errors.package$.attachTree(package.scala:49)
      at org.apache.spark.sql.execution.aggregate.TungstenAggregate.doExecute(TungstenAggregate.scala:69)
      at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$5.apply(SparkPlan.scala:140)
      at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$5.apply(SparkPlan.scala:138)
      at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:147)
      at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:138)
      at org.apache.spark.sql.execution.SparkPlan.executeCollect(SparkPlan.scala:174)
      at org.apache.spark.sql.DataFrame$$anonfun$collect$1.apply(DataFrame.scala:1386)
      at org.apache.spark.sql.DataFrame$$anonfun$collect$1.apply(DataFrame.scala:1386)
      at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:56)
      at org.apache.spark.sql.DataFrame.withNewExecutionId(DataFrame.scala:1904)
      at org.apache.spark.sql.DataFrame.collect(DataFrame.scala:1385)
      at org.apache.spark.sql.DataFrame.count(DataFrame.scala:1403)
      at com.salesforce.hammer.clusterer.collector.DeduperTest$.applyBaselineAndUpgradeOnlyPatternTest(DeduperTest.scala:83)
      at com.salesforce.hammer.clusterer.collector.Application$.main(Main.scala:53)
      at com.salesforce.hammer.clusterer.collector.Application.main(Main.scala)
  Caused by: org.apache.spark.sql.catalyst.errors.package$TreeNodeException: execute, tree:
  TungstenExchange SinglePartition
   TungstenAggregate(key=[], functions=[(count(1),mode=Partial,isDistinct=false)], output=[currentCount#121L])
    Union
     TungstenProject
       Filter UDF(Col1Text#97,Col2Text#109)
        Scan CsvRelation(playWithSpark/data/dataDf1.csv,true,,,",null,#,PERMISSIVE,COMMONS,false,false,null,UTF-8,false)[Col1Text#97,BaselineStacktrace#98,BaselineTime#99,ClassId#100,ClassName#101,Id#102,IsDataSilo#103,MethodName#104,Namespace#105,Organization#106,PackageName#107,TestResultKey#108,Col2Text#109,UpgradedStacktrace#110,UpgradedTime#111]
      TungstenProject
       Filter UDF(Col1Text#2,Col2Text#14)
        Scan CsvRelation(playWithSpark/data/dataDf2.csv,true,,,",null,#,PERMISSIVE,COMMONS,false,false,null,UTF-8,false)[Col1Text#2,BaselineStacktrace#3,BaselineTime#4,ClassId#5,ClassName#6,Id#7,IsDataSilo#8,MethodName#9,Namespace#10,Organization#11,PackageName#12,TestResultKey#13,Col2Text#14,UpgradedStacktrace#15,UpgradedTime#16]
      at org.apache.spark.sql.catalyst.errors.package$.attachTree(package.scala:49)
      at org.apache.spark.sql.execution.Exchange.doExecute(Exchange.scala:141)
      at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$5.apply(SparkPlan.scala:140)
      at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$5.apply(SparkPlan.scala:138)
      at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:147)
      at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:138)
      at org.apache.spark.sql.execution.aggregate.TungstenAggregate$$anonfun$doExecute$1.apply(TungstenAggregate.scala:119)
      at org.apache.spark.sql.execution.aggregate.TungstenAggregate$$anonfun$doExecute$1.apply(TungstenAggregate.scala:69)
      at org.apache.spark.sql.catalyst.errors.package$.attachTree(package.scala:48)
      ... 15 more
  Caused by: org.apache.spark.sql.catalyst.errors.package$TreeNodeException: execute, tree:
  TungstenAggregate(key=[], functions=[(count(1),mode=Partial,isDistinct=false)], output=[currentCount#121L])
   Union
  
  TungstenProject
       Filter UDF(Col1Text#97,Col2Text#109)
        Scan CsvRelation(playWithSpark/data/dataDf1.csv,true,,,",null,#,PERMISSIVE,COMMONS,false,false,null,UTF-8,false)[Col1Text#97,BaselineStacktrace#98,BaselineTime#99,ClassId#100,ClassName#101,Id#102,IsDataSilo#103,MethodName#104,Namespace#105,Organization#106,PackageName#107,TestResultKey#108,Col2Text#109,UpgradedStacktrace#110,UpgradedTime#111]
      TungstenProject
       Filter UDF(Col1Text#2,Col22Text#14)
        Scan CsvRelation(playWithSpark/data/dataDf2.csv,true,,,",null,#,PERMISSIVE,COMMONS,false,false,null,UTF-8,false)[Col1Text#2,BaselineStacktrace#3,BaselineTime#4,ClassId#5,ClassName#6,Id#7,IsDataSilo#8,MethodName#9,Namespace#10,Organization#11,PackageName#12,TestResultKey#13,Col2Text#14,UpgradedStacktrace#15,UpgradedTime#16]
  
  
      at org.apache.spark.sql.catalyst.errors.package$.attachTree(package.scala:49)
      at org.apache.spark.sql.execution.aggregate.TungstenAggregate.doExecute(TungstenAggregate.scala:69)
      at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$5.apply(SparkPlan.scala:140)
      at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$5.apply(SparkPlan.scala:138)
      at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:147)
      at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:138)
      at org.apache.spark.sql.execution.Exchange$$anonfun$doExecute$1.apply(Exchange.scala:142)
      at org.apache.spark.sql.execution.Exchange$$anonfun$doExecute$1.apply(Exchange.scala:141)
      at org.apache.spark.sql.catalyst.errors.package$.attachTree(package.scala:48)
      ... 23 more
  Caused by: 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:2021)
      at org.apache.spark.rdd.RDD$$anonfun$mapPartitions$1.apply(RDD.scala:703)
      at org.apache.spark.rdd.RDD$$anonfun$mapPartitions$1.apply(RDD.scala:702)
      at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:147)
      at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:108)
      at org.apache.spark.rdd.RDD.withScope(RDD.scala:306)
      at org.apache.spark.rdd.RDD.mapPartitions(RDD.scala:702)
      at org.apache.spark.sql.execution.Filter.doExecute(basicOperators.scala:113)
      at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$5.apply(SparkPlan.scala:140)
      at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$5.apply(SparkPlan.scala:138)
      at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:147)
      at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:138)
      at org.apache.spark.sql.execution.TungstenProject.doExecute(basicOperators.scala:86)
      at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$5.apply(SparkPlan.scala:140)
      at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$5.apply(SparkPlan.scala:138)
      at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:147)
      at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:138)
      at org.apache.spark.sql.execution.Union$$anonfun$doExecute$1.apply(basicOperators.scala:184)
      at org.apache.spark.sql.execution.Union$$anonfun$doExecute$1.apply(basicOperators.scala:184)
      at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
      at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
      at scala.collection.immutable.List.foreach(List.scala:309)
      at scala.collection.TraversableLike$class.map(TraversableLike.scala:244)
      at scala.collection.AbstractTraversable.map(Traversable.scala:105)
      at org.apache.spark.sql.execution.Union.doExecute(basicOperators.scala:184)
      at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$5.apply(SparkPlan.scala:140)
      at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$5.apply(SparkPlan.scala:138)
      at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:147)
      at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:138)
      at org.apache.spark.sql.execution.aggregate.TungstenAggregate$$anonfun$doExecute$1.apply(TungstenAggregate.scala:119)
      at org.apache.spark.sql.execution.aggregate.TungstenAggregate$$anonfun$doExecute$1.apply(TungstenAggregate.scala:69)
      at org.apache.spark.sql.catalyst.errors.package$.attachTree(package.scala:48)
      ... 31 more
  Caused by: java.io.NotSerializableException: com.salesforce.RegexDeduper
  Serialization stack:
      - object not serializable (class: com.salesforce.RegexDeduper, value: com.salesforce.RegexDeduper@67b464f4)
      - field (class: com.salesforce.RegexDeduper$$anonfun$applyCol1TextAndCol2TextPattern$1, name: $outer, type: class com.salesforce.RegexDeduper)
      - object (class com.salesforce.RegexDeduper$$anonfun$applyBaselineAndUpgradePattern$1, &lt;function2&gt;)
      - field (class: org.apache.spark.sql.catalyst.expressions.ScalaUDF$$anonfun$3, name: func$3, type: interface scala.Function2)
      - object (class org.apache.spark.sql.catalyst.expressions.ScalaUDF$$anonfun$3, &lt;function1&gt;)
      - field (class: org.apache.spark.sql.catalyst.expressions.ScalaUDF, name: f, type: interface scala.Function1)
      - object (class org.apache.spark.sql.catalyst.expressions.ScalaUDF, UDF(Col1Text#97,Col2Text#109))
      - field (class: org.apache.spark.sql.execution.Filter, name: condition, type: class org.apache.spark.sql.catalyst.expressions.Expression)
      - object (class org.apache.spark.sql.execution.Filter, Filter UDF(Col1Text#97,Col2Text#109)
   Scan CsvRelation(playWithSpark/data/dataDf1.csv,true,,,",null,#,PERMISSIVE,COMMONS,false,false,null,UTF-8,false)[Col1Text#97,BaselineStacktrace#98,BaselineTime#99,ClassId#100,ClassName#101,Id#102,IsDataSilo#103,MethodName#104,Namespace#105,Organization#106,PackageName#107,TestResultKey#108,Col2TText#109,UpgradedStacktrace#110,UpgradedTime#111]
  )
      - field (class: org.apache.spark.sql.execution.Filter$$anonfun$4, name: $outer, type: class org.apache.spark.sql.execution.Filter)
      - object (class org.apache.spark.sql.execution.Filter$$anonfun$4, &lt;function1&gt;)
      at org.apache.spark.serializer.SerializationDebugger$.improveException(SerializationDebugger.scala:40)
      at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:47)
      at org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:84)
      at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:301)
  </code></pre>