title: java.io.NotSerializableException: org.apache.spark.sql.Column
body: <p>I have a data frame with three columns. I have to perform a mapping on last recent k records of the dataframe based on a groupBy condition. Here is the code that I wrote and it is throwing NotSerializable Exception The dataframe looks like below I have to group by them using cities and have to compute value on the last two records for every city. </p>
  
  <pre><code>&lt;pre&gt;&lt;code&gt;
      +------------+-----+-----+
      |      cities|temps|preas|
      +------------+-----+-----+
      |   Hyderabad|   30|    3|
      |Secunderabad|   38|    3|
      |      Mumbai|   41|    4|
      |     Chennai|   39|    3|
      |      Mumbai|   25|    2|
      |     Chennai|   21|    2|
      |   Hyderabad|   36|    3|
      |secunderabad|   32|    3| 
      |     Chennai|   26|    2| 
      |      Mumbai|   45|    4|
      |     Chennai|   20|    2|
      |   Hyderabad|   39|    3|
      |Secunderabad|   42|    4|
      +------------+-----+-----+
  
  `   import org.apache.spark.sql.expressions.Window
      val k = 2
      @transient val w = Window.partitionBy("cities") 
      val coll = List("cities","temps","preas")
      val cl = coll.map(a =&gt; col(a))
      val colls = cl :+ row_number.over(w).as("row_number")
      @transient val nums = df.select(colls:_*)
      val limits 
      =nums.groupBy("cities").agg(max($"row_number").minus(k).as("limit"))
      val data = nums.join(limits, "cities").filter($"row_number".gt($"limit"))
      val 
      dfd=data.withColumn("row_number",$"row_number".minus($"limit")).select(cl:_*)
      val dfRDD = dfd.rdd
      import org.apache.spark.sql.Row
      val weights = List(1,2)
      val func = (it:Iterable[Row]) =&gt; {
          val lis = it.toList
          var res :List[Int] = List()
          val len = lis(0).size
          for (i&lt;-1 until len){
            res = res:+lis.map(a=&gt;a(i)).map(b=&gt;b.asInstanceOf[Int]).zip(weights).map(c=&gt;c._1*c._2).sum/(weights.reduce(_+_))
          }
          res
      }
      val outp = dfRDD.groupBy(row =&gt; row(0)).map{case(k,v)=&gt;(k,func(v))`
  </code></pre>