title: Problems with adding a new column to a dataframe - spark/scala
body: <p>I am new to spark/scala. I am trying to read some data from a hive table to a spark dataframe and then add a column based on some condition. Here is my code:</p>
  
  <pre><code>val DF = hiveContext.sql("select * from (select * from test_table where partition_date='2017-11-22') a JOIN (select max(id) as bid from test_table where partition_date='2017-11-22' group by at_id) b ON a.id=b.bid")
  
  def dateDiff(partition_date: org.apache.spark.sql.Column, item_due_date: org.apache.spark.sql.Column): Long ={
        ChronoUnit.DAYS.between(LocalDate.parse(partition_date.toString()), LocalDate.parse(item_due_date.toString))
      }
  
  val finalDF = DF.withColumn("status", 
                     when(col("past_due").equalTo(1) &amp;&amp; !(col("item_due_date").equalTo(null) || col("item_due_date").equalTo("NULL") || col("item_due_date").equalTo("null")) &amp;&amp; (dateDiff(col("partition_date"),col("item_due_date")) &lt; 0) &amp;&amp; !(col("item_decision").equalTo(null) || col("item_decision").equalTo("NULL") || col("item_decision").equalTo("null")), "approved")
                    .when(col("past_due").equalTo(1) &amp;&amp; !(col("item_due_date").equalTo(null) || col("item_due_date").equalTo("NULL") || col("item_due_date").equalTo("null")) &amp;&amp; (dateDiff(col("partition_date"),col("item_due_date")) &lt; 0) &amp;&amp; (col("item_decision").equalTo(null) || col("item_decision").equalTo("NULL") || col("item_decision").equalTo("null")), "pending")
                    .when(col("past_due").equalTo(1) &amp;&amp; !(col("item_due_date").equalTo(null) || col("item_due_date").equalTo("NULL") || col("item_due_date").equalTo("null")) &amp;&amp; (dateDiff(col("partition_date"),col("item_due_date")) &gt;= 0), "expired")
                    .otherwise("null"))
  </code></pre>
  
  <p><code>dateDiff</code> is a function that calculates the difference between <code>partition_date</code> and <code>item_due_date</code>, which are columns in <code>DF</code>. I am trying to add a new column to <code>DF</code> by using <code>when</code> and <code>otherwise</code> which uses the <code>dateDiff</code> to get the difference between dates. </p>
  
  <p>Now, when I run the above code, I get the following error:  <code>org.threeten.bp.format.DateTimeParseException: Text 'partition_date' could not be parsed at index 0</code> </p>
  
  <p>I believe the value of the column <code>partition_date</code> is not being converted to a String to be parsed as a date. Is this what's happening? If yes, how do I cast the column value to a String ?</p>
  
  <p>Below is the schema of the columns I am using from the <code>DF</code> :</p>
  
  <pre><code> |-- item_due_date: string (nullable = true)
   |-- past_due: integer (nullable = true)
   |-- item_decision: string (nullable = true)
   |-- partition_date: string (nullable = true)
  </code></pre>
  
  <p>A data sample of the columns I am using from the <code>DF</code> :</p>
  
  <pre><code>+--------+-------------+-------------+--------------+
  |past_due|item_due_date|item_decision|partition_date|
  +--------+-------------+-------------+--------------+
  |       1|   0001-01-14|         null|    2017-11-22|
  |       1|   0001-01-14|     Mitigate|    2017-11-22|
  |       1|   0001-01-14|     Mitigate|    2017-11-22|
  |       1|   0001-01-14|     Mitigate|    2017-11-22|
  |       0|   2018-03-18|         null|    2017-11-22|
  |       1|   2016-11-30|         null|    2017-11-22|
  +--------+-------------+-------------+--------------+
  </code></pre>
  
  <p>I also tried using a custom UDF:</p>
  
  <pre><code>  def status(past_due: Int, item_decision: String, maxPartitionDate: String, item_due_date: String): String = {
        if (past_due == 1 &amp;&amp; item_due_date != "NULL") {
          if (ChronoUnit.DAYS.between(LocalDate.parse(maxPartitionDate.trim), LocalDate.parse(item_due_date.trim)) &lt; 0) {
            if (item_decision != "NULL") "pending"
            else "approved"
          } else "expired"
        } else "NULL"
      }
  
  val statusUDF = sqlContext.udf.register("statusUDF", status _)
  
  val DF2 = DF.withColumn("status", statusUDF(DF("past_due"),DF("item_decision"),DF("partition_date"),DF("item_due_date")))
  DF2.show()
  </code></pre>
  
  <p>And it throws the following error at the <code>DF2.show</code> statement, everytime:</p>
  
  <pre><code>Container exited with a non-zero exit code 50
  
  Driver stacktrace:
          at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1433)
          at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1421)
          at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1420)
          at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
          at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)
          at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1420)
          at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:799)
          at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:799)
          at scala.Option.foreach(Option.scala:236)
          at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:799)
          at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:1644)
          at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1603)
          at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1592)
          at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48)
          at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:620)
          at org.apache.spark.SparkContext.runJob(SparkContext.scala:1844)
          at org.apache.spark.SparkContext.runJob(SparkContext.scala:1857)
          at org.apache.spark.SparkContext.runJob(SparkContext.scala:1870)
          at org.apache.spark.sql.execution.SparkPlan.executeTake(SparkPlan.scala:212)
          at org.apache.spark.sql.execution.Limit.executeCollect(basicOperators.scala:165)
          at org.apache.spark.sql.execution.SparkPlan.executeCollectPublic(SparkPlan.scala:174)
          at org.apache.spark.sql.DataFrame$$anonfun$org$apache$spark$sql$DataFrame$$execute$1$1.apply(DataFrame.scala:1499)
          at org.apache.spark.sql.DataFrame$$anonfun$org$apache$spark$sql$DataFrame$$execute$1$1.apply(DataFrame.scala:1499)
          at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:53)
          at org.apache.spark.sql.DataFrame.withNewExecutionId(DataFrame.scala:2086)
          at org.apache.spark.sql.DataFrame.org$apache$spark$sql$DataFrame$$execute$1(DataFrame.scala:1498)
          at org.apache.spark.sql.DataFrame.org$apache$spark$sql$DataFrame$$collect(DataFrame.scala:1505)
          at org.apache.spark.sql.DataFrame$$anonfun$head$1.apply(DataFrame.scala:1375)
          at org.apache.spark.sql.DataFrame$$anonfun$head$1.apply(DataFrame.scala:1374)
          at org.apache.spark.sql.DataFrame.withCallback(DataFrame.scala:2099)
          at org.apache.spark.sql.DataFrame.head(DataFrame.scala:1374)
          at org.apache.spark.sql.DataFrame.take(DataFrame.scala:1456)
          at org.apache.spark.sql.DataFrame.showString(DataFrame.scala:170)
          at org.apache.spark.sql.DataFrame.show(DataFrame.scala:350)
          at org.apache.spark.sql.DataFrame.show(DataFrame.scala:311)
          at org.apache.spark.sql.DataFrame.show(DataFrame.scala:319)
          at driver$.main(driver.scala:109)
          at driver.main(driver.scala)
  </code></pre>
  
  <p>Any help would be appreciated. Thank you!</p>