title: Spark UserDefinedAggregateFunction: scala.MatchError 0.0 (of class java.lang.Double)
body: <p>I'm trying to use the <code>UserDefinedAggregateFunction</code> on Spark 2.0.2 with Scala, but I'm coming across a match error. I've created the below as a test case, the code that I'm writing does something similar to the below.</p>
  
  <p>I am trying to accumulate a value through an aggregation window. It's not just a cumulative sum, but I need to calculate the number to keep, based on some conditions.</p>
  
  <p>As a test case, I have created an amortization table, where I have to calculate the opening and closing balance of each month.</p>
  
  <p>The data looks like:</p>
  
  <pre class="lang-none prettyprint-override"><code>+------+--------+------------+---------+
  |Period| Capital|InterestRate|Repayment|
  +------+--------+------------+---------+
  |201601|   0.00 |       0.10 |    0.00 |
  |201602|1000.00 |       0.00 |    0.00 |
  |201603|2000.00 |       0.10 |    0.00 |
  |201604|   0.00 |       0.10 | -200.00 |
  |201605|   0.00 |       0.10 | -200.00 |
  |201606|   0.00 |       0.10 | -200.00 |
  |201607|   0.00 |       0.10 | -200.00 |
  |201608|   0.00 |       0.00 | -200.00 |
  |201609|   0.00 |       0.10 | -200.00 |
  |201610|   0.00 |       0.10 | -200.00 |
  |201611|   0.00 |       0.10 | -200.00 |
  |201612|   0.00 |       0.10 | -200.00 |
  +------+--------+------------+---------+
  </code></pre>
  
  <p>I couldn't get the CSV to format properly, but I've added it to a gist here: <a href="https://gist.github.com/nevi-me/8b2362a5365e73af947fc13bb5836adc" rel="nofollow noreferrer">https://gist.github.com/nevi-me/8b2362a5365e73af947fc13bb5836adc</a>.</p>
  
  <p>I'm trying to calculate the <code>Opening</code> and <code>Closing</code> balances, and then return the <code>Closing</code> balance from the aggregation.</p>
  
  <h1>Scala</h1>
  
  <pre class="lang-scala prettyprint-override"><code>package me.nevi
  
  import org.apache.spark.sql._
  import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction, Window}
  import org.apache.spark.sql.types.{StructType, DoubleType, DataType}
  
  object AggregationTest {
  
    object amortisedClosingBalance extends UserDefinedAggregateFunction {
      override def inputSchema: StructType = new StructType().add("Capital", DoubleType).add("InterestRate", DoubleType).add("Repayment", DoubleType)
  
      override def bufferSchema: StructType = new StructType().add("Opening", DoubleType).add("Closing", DoubleType)
  
      override def dataType: DataType = new StructType().add("Closing", DoubleType)
  
      override def deterministic: Boolean = true
  
      override def initialize(buffer: MutableAggregationBuffer): Unit = {
        buffer.update(0, 0.0)
        buffer.update(1, 0.0)
      }
  
      override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
        if (!input.isNullAt(0)) {
          println(buffer.get(0))
          println(buffer.get(1))
          buffer.update(0, buffer.getDouble(1))
          // (opening + capital) * interestrate - repayment
          buffer.update(1, (buffer.getDouble(0) + input.getDouble(0)) * input.getDouble(1) + input.getDouble(2))
        } else {
          // if first record?
          buffer.update(0, input.getDouble(0))
          buffer.update(1, input.getDouble(0))
        }
      }
  
      override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
        buffer1.update(0, buffer1.getDouble(0))
        buffer1.update(1, buffer1.getDouble(1))
      }
  
      override def evaluate(buffer: Row): Any = {
        buffer.getDouble(1)
      }
    }
  
    def main(args: Array[String]): Unit = {
      System.setProperty("hadoop.home.dir", "C:/spark")
      System.setProperty("spark.sql.warehouse.dir", "file:///tmp/spark-warehouse")
  
      val spark: SparkSession = SparkSession.builder()
        .master("local[*]")
        .appName("Aggregation Test")
        .getOrCreate()
  
      import spark.implicits._
  
      val df = spark.read.option("header", true).csv("file:///d:/interest_calc.csv")
  
      df.show()
  
      val windowSpec = Window.orderBy(df.col("Period"))
  
      val calc = df.withColumn("Closing", amortisedClosingBalance($"Capital", $"InterestRate", $"Repayment").over(windowSpec))
  
      calc.show()
  
    }
  }
  </code></pre>
  
  <p>I get the exception:</p>
  
  <pre class="lang-none prettyprint-override"><code>scala.MatchError: 0.0 (of class java.lang.Double)
    at org.apache.spark.sql.catalyst.CatalystTypeConverters$StructConverter.toCatalystImpl(CatalystTypeConverters.scala:256)
    at org.apache.spark.sql.catalyst.CatalystTypeConverters$StructConverter.toCatalystImpl(CatalystTypeConverters.scala:251)
    at org.apache.spark.sql.catalyst.CatalystTypeConverters$CatalystTypeConverter.toCatalyst(CatalystTypeConverters.scala:103)
    at org.apache.spark.sql.catalyst.CatalystTypeConverters$$anonfun$createToCatalystConverter$2.apply(CatalystTypeConverters.scala:403)
    at org.apache.spark.sql.execution.aggregate.ScalaUDAF.eval(udaf.scala:440)
    at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificMutableProjection.apply(Unknown Source)
    at org.apache.spark.sql.execution.AggregateProcessor.evaluate(WindowExec.scala:1029)
    at org.apache.spark.sql.execution.UnboundedPrecedingWindowFunctionFrame.write(WindowExec.scala:822)
    at org.apache.spark.sql.execution.WindowExec$$anonfun$15$$anon$1.next(WindowExec.scala:398)
    at org.apache.spark.sql.execution.WindowExec$$anonfun$15$$anon$1.next(WindowExec.scala:289)
    at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIterator.processNext(Unknown Source)
    at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
    at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$8$$anon$1.hasNext(WholeStageCodegenExec.scala:370)
    at org.apache.spark.sql.execution.SparkPlan$$anonfun$4.apply(SparkPlan.scala:246)
    at org.apache.spark.sql.execution.SparkPlan$$anonfun$4.apply(SparkPlan.scala:240)
    at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$24.apply(RDD.scala:803)
    at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$24.apply(RDD.scala:803)
    at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:319)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:283)
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:70)
    at org.apache.spark.scheduler.Task.run(Task.scala:86)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:274)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
  </code></pre>
  
  <p>Does anyone know what I am doing wrong? I initially used Spark 2.0.0, and I came across someone else with a similar issue with UDTF's, and the suggestion was to upgrade to 2.0.1, however after upgrading; my issue still persists.</p>
  
  <hr>
  
  <h1>Solution: for anyone interested</h1>
  
  <p>The problem was with my schema as per the accepted answer. The below is the snippet with the calculations working correctly. </p>
  
  <pre><code>package me.nevi
  
  import org.apache.spark.sql._
  import org.apache.spark.sql.functions._
  import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction, Window}
  import org.apache.spark.sql.types.{DataType, DoubleType, StructType}
  
  object AggregationTest {
  
    object amortisedClosingBalance extends UserDefinedAggregateFunction {
      override def inputSchema: StructType = new StructType().add("Capital", DoubleType).add("InterestRate", DoubleType).add("Repayment", DoubleType)
  
      override def bufferSchema: StructType = new StructType().add("Opening", DoubleType).add("Closing", DoubleType)
  
      override def dataType: DataType = new StructType().add("Opening", DoubleType).add("Closing", DoubleType)
  
      override def deterministic: Boolean = true
  
      override def initialize(buffer: MutableAggregationBuffer): Unit = {
        buffer.update(0, 0.0)
        buffer.update(1, 0.0)
      }
  
      override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
        if (!input.isNullAt(0)) {
          println(buffer.get(0))
          println(buffer.get(1))
          buffer.update(0, buffer.getDouble(1))
          // (opening + capital) * interestrate - repayment
          buffer.update(1, input.getDouble(0)
            + buffer.getDouble(0) + input.getDouble(2) + (buffer.getDouble(0) + input.getDouble(0)) * (input.getDouble(1) / 12))
        } else {
          // if first record?
          buffer.update(0, input.getDouble(0))
          buffer.update(1, input.getDouble(0))
        }
      }
  
      override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
        buffer1.update(0, buffer1.getDouble(0))
        buffer1.update(1, buffer1.getDouble(1))
      }
  
      override def evaluate(buffer: Row): Any = {
        Row(buffer.getDouble(0), buffer.getDouble(1))
      }
    }
  
    def main(args: Array[String]): Unit = {
      System.setProperty("hadoop.home.dir", "C:/spark")
      System.setProperty("spark.sql.warehouse.dir", "file:///tmp/spark-warehouse")
  
      val spark: SparkSession = SparkSession.builder()
        .master("local[*]")
        .appName("Aggregation Test")
        .getOrCreate()
  
      import spark.implicits._
  
      val df = spark.read.option("header", true).csv("file:///d:/interest_calc.csv")
  
      df.show()
  
      val windowSpec = Window.orderBy(df.col("Period").asc)
  
      var calc = df.withColumn("Calcs", amortisedClosingBalance($"Capital", $"InterestRate", $"Repayment").over(windowSpec))
      calc = calc.withColumn("Opening", round($"Calcs".getField("Opening"), 2)).withColumn("Closing", round($"Calcs".getField("Closing"),2))
        .drop("Calcs")
  
      calc.show()
  
    }
  }
  </code></pre>
  
  <p>With the below result:</p>
  
  <pre class="lang-none prettyprint-override"><code>+------+--------+------------+---------+-------+-------+
  |Period| Capital|InterestRate|Repayment|Opening|Closing|
  +------+--------+------------+---------+-------+-------+
  |201601|   0.00 |       0.10 |    0.00 |    0.0|    0.0|
  |201602|1000.00 |       0.00 |    0.00 |    0.0| 1000.0|
  |201603|2000.00 |       0.10 |    0.00 | 1000.0| 3025.0|
  |201604|   0.00 |       0.10 | -200.00 | 3025.0|2850.21|
  |201605|   0.00 |       0.10 | -200.00 |2850.21|2673.96|
  |201606|   0.00 |       0.10 | -200.00 |2673.96|2496.24|
  |201607|   0.00 |       0.10 | -200.00 |2496.24|2317.05|
  |201608|   0.00 |       0.00 | -200.00 |2317.05|2117.05|
  |201609|   0.00 |       0.10 | -200.00 |2117.05|1934.69|
  |201610|   0.00 |       0.10 | -200.00 |1934.69|1750.81|
  |201611|   0.00 |       0.10 | -200.00 |1750.81| 1565.4|
  |201612|   0.00 |       0.10 | -200.00 | 1565.4|1378.44|
  +------+--------+------------+---------+-------+-------+
  </code></pre>