Spark-sql error by defining UDAF

Question

Spark version 1.6.0 on AWS EMR with Zeppelin Notebook

I defined a UDAF with following codes:

import org.apache.spark.sql.types._
import org.apache.spark.sql.Row;
import org.apache.spark.sql.expressions.MutableAggregationBuffer
import org.apache.spark.sql.expressions.UserDefinedAggregateFunction

import java.text.SimpleDateFormat
import java.util.Date

class AggregateTS extends UserDefinedAggregateFunction{
    def inputSchema: StructType = StructType(StructField("input", StringType) :: Nil)

    def bufferSchema: StructType = StructType(StructField("intermediate", StringType)::Nil)

    def dataType: DataType = StringType

    def deterministic: Boolean = true

    def initialize(buffer: MutableAggregationBuffer): Unit = {
        buffer(0) = "Init"
    }

    def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
        if (buffer.getAs[String](0) == "Init"){
            buffer(0) = input.getAs[String](0)
        }
        else{
            // add two string
            buffer(0) = average_ts(input.getAs[String](0), buffer.getAs[String](0))
        }
    }

    def merge(buffer1: MutableAggregationBuffer, buffer2:Row):Unit = {
        buffer1(0) = average_ts(buffer1.getAs[String](0), buffer2.getAs[String](0))
    }

    def evaluate(buffer: Row): Any = {
        buffer.getAs[String](0)
    }
}

from which I got a compile error:

error: not found: type DataType
       def dataType: DataType = StringType

What does it mean ?

Tried your code - and it compiles :| which Spark version are you using? I compiled it with 1.5.2 — Tzach Zohar

Hello lad Hello lad · Accepted Answer · 2016-03-08T13:42:02

solved by myself. It seems to be some import collision error. I change the import sentence to explicit

import org.apache.spark.sql.types.{DataType}

it works then

Spark-sql error by defining UDAF

1 Answers