title: Exception when using a udf function with Spark DataFrame
body: <p>In Spark version: 2.4.0, I am trying to execute the code below on the given DataFrame:
  unfoldedDF:org.apache.spark.sql.DataFrame<br>
  movieid:integer<br>
  words:array -- element:string<br>
  tokens:string</p>
  
  <pre class="lang-scala prettyprint-override"><code>val tokensWithDf = unfoldedDF.groupBy("tokens").agg(countDistinct("movieid") as "df")
  tokensWithDf.show()
  </code></pre>
  
  <p>The new dataframe created is tokensWithDf:org.apache.spark.sql.DataFrame<br>
  tokens:string<br>
  df:long</p>
  
  <p>On it the following operation is done.</p>
  
  <pre class="lang-scala prettyprint-override"><code>def findIdf(x : Long) : Double = {scala.math.log10((42306).toDouble/x)}
  val sqlfunc = udf(findIdf _)
  tokensWithDf.withColumn("idf", sqlfunc(col("df"))).show()
  </code></pre>
  
  <p>It fails with the following exception:  </p>
  
  <pre><code>org.apache.spark.SparkException: Task not serializable
      at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:403)
      at org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:393)
      at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:162)
      at org.apache.spark.SparkContext.clean(SparkContext.scala:2519)
      at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsWithIndex$1.apply(RDD.scala:866)
      at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsWithIndex$1.apply(RDD.scala:865)
      at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
      at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
      at org.apache.spark.rdd.RDD.withScope(RDD.scala:379)
      at org.apache.spark.rdd.RDD.mapPartitionsWithIndex(RDD.scala:865)
      at org.apache.spark.sql.execution.WholeStageCodegenExec.doExecute(WholeStageCodegenExec.scala:616)
      at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:143)
      at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:131)
      at org.apache.spark.sql.execution.SparkPlan$$anonfun$executeQuery$5.apply(SparkPlan.scala:183)
      at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
      at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:180)
      at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:131)
      at org.apache.spark.sql.execution.collect.Collector$.collect(Collector.scala:66)
      at org.apache.spark.sql.execution.collect.Collector$.collect(Collector.scala:75)
      at org.apache.spark.sql.execution.ResultCacheManager.getOrComputeResult(ResultCacheManager.scala:497)
      at org.apache.spark.sql.execution.CollectLimitExec.executeCollectResult(limit.scala:48)
  </code></pre>