My data lab

val td = rd.select($"jobid".alias("jobid"), $"time".alias("starttime"), $"time".alias("endtime"), $"result".alias("report_result"), $"result".alias("error"))
//val td = rd.select($"jobid".alias("jobid"),$"time".alias("starttime"))

import CustomAgg._
val minagg = new minAggregation()
val maxagg = new maxAggregation()
val erragg = new errorAggregation()

val td2 = td.groupBy("jobid").agg(minagg(td("starttime")), maxagg(td("endtime")) , erragg(td("error"))) 

ackage CustomAgg

import org.apache.spark.sql.Row
import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction}
import org.apache.spark.sql.types._
import org.apache.spark.unsafe.types.UTF8String

import scala.collection.mutable.ArrayBuffer

object CustomAggregation
  {

    class maxAggregation extends UserDefinedAggregateFunction {
    // Input Data Type Schema
    //def inputSchema: StructType = StructType(Array(StructField("col5", StringType)))

    // Intermediate Schema
    //def bufferSchema = StructType(Array(    StructField("col5_collapsed", StringType)))

      def inputSchema = new StructType().add("x", StringType)
      def bufferSchema = new StructType().add("buff", ArrayType(StringType))
    // Returned Data Type .
    def dataType: DataType = StringType

    // Self-explaining
    def deterministic = true

    // This function is called whenever key changes
    def initialize(buffer: MutableAggregationBuffer) = {
      buffer.update(0, ArrayBuffer.empty[String])// initialize array
    }

    // Iterate over each entry of a group
    def update(buffer: MutableAggregationBuffer, input: Row) = {
      if (!input.isNullAt(0)) {
        println(input.getString(0))
        println(buffer.getSeq[String](0).mkString)
        if(buffer.getSeq[String](0).mkString.isEmpty)
          buffer.update(0, Seq(input.getString(0)))
        else if(buffer.getSeq[String](0).mkString.compareTo(input.getString(0)) > 0)
          buffer.update(0, buffer.getSeq[String](0))
        else
          buffer.update(0, Seq(input.getString(0)))
        println(buffer.getSeq[String](0).mkString)
      }
    }


    // Merge two partial aggregates
    def merge(buffer1: MutableAggregationBuffer, buffer2: Row) = {
      if(buffer1.getSeq[String](0).mkString.compareTo(buffer2.getSeq[String](0).mkString) > 0)
        buffer1.update(0, buffer1.getSeq[String](0))
      else
        buffer1.update(0, buffer2.getSeq[String](0))
    }

      def evaluate(buffer: Row) = UTF8String.fromString(
        buffer.getSeq[String](0).mkString(","))
  }

    class minAggregation extends UserDefinedAggregateFunction {
      // Input Data Type Schema
      //def inputSchema: StructType = StructType(Array(StructField("col5", StringType)))

      // Intermediate Schema
      //def bufferSchema = StructType(Array(    StructField("col5_collapsed", StringType)))

      def inputSchema = new StructType().add("x", StringType)
      def bufferSchema = new StructType().add("buff",StringType)
      // Returned Data Type .
      def dataType: DataType = StringType

      // Self-explaining
      def deterministic = true

      // This function is called whenever key changes
      def initialize(buffer: MutableAggregationBuffer) = {
        buffer.update(0, "")// initialize array
      }

      // Iterate over each entry of a group
      def update(buffer: MutableAggregationBuffer, input: Row) = {
        if (!input.isNullAt(0)) {

          if(buffer.getString(0).isEmpty)
            buffer.update(0, input.getString(0))
          else if(buffer.getString(0).compareTo(input.getString(0)) < 0)
            buffer.update(0, buffer.getString(0))
          else
            buffer.update(0, input.getString(0))

          println( "updated :" + buffer.getString(0))
        }
      }


      // Merge two partial aggregates
      def merge(buffer1: MutableAggregationBuffer, buffer2: Row) = {

        if(buffer1.getString(0).compareTo(buffer2.getString(0))< 0 && !buffer1.getString(0).isEmpty)
          buffer1.update(0, buffer1.getString(0))
        else
          buffer1.update(0, buffer2.getString(0))
      }

      def evaluate(buffer: Row) = UTF8String.fromString(
        buffer.getString(0))
    }


    class errorAggregation extends UserDefinedAggregateFunction {
      // Input Data Type Schema
      //def inputSchema: StructType = StructType(Array(StructField("col5", StringType)))

      // Intermediate Schema
      //def bufferSchema = StructType(Array(    StructField("col5_collapsed", StringType)))

      def inputSchema = new StructType().add("x", StringType)
      def bufferSchema = new StructType().add("buff", ArrayType(StringType))
      // Returned Data Type .
      def dataType: DataType = StringType

      // Self-explaining
      def deterministic = true

      // This function is called whenever key changes
      def initialize(buffer: MutableAggregationBuffer) = {
        buffer.update(0, Seq(""))// initialize array
      }

      def checkerror (str : String) : Boolean = {

        if(str.compareTo("Info") == 0) false
        if(str.compareTo("0") == 0) false
        if(str.compareTo("Success") == 0) false
        if(str.compareTo("") == 0) false
        true
      }

      // Iterate over each entry of a group
      def update(buffer: MutableAggregationBuffer, input: Row) = {
        if (!input.isNullAt(0)) {
          if(checkerror(input.getString(0)))
          {
            input.getString(0)
            buffer.update(0, Seq(input.getString(0)))
          }
        }
      }


      // Merge two partial aggregates
      def merge(buffer1: MutableAggregationBuffer, buffer2: Row) = {
        println(buffer1.getSeq[String](0).toString())
        println(buffer2.getSeq[String](0).toString())
        buffer1.update(0, buffer1.getSeq[String](0) ++ buffer2.getSeq[String](0))
      }

      def evaluate(buffer: Row) = UTF8String.fromString(
        buffer.getSeq[String](0).mkString(","))
    }


    class GeometricMean extends UserDefinedAggregateFunction {
      // This is the input fields for your aggregate function.
      override def inputSchema: org.apache.spark.sql.types.StructType =
        StructType(StructField("value", StringType) :: Nil)

      // This is the internal fields you keep for computing your aggregate.
      override def bufferSchema: StructType = StructType(
        StructField("count", LongType) ::
          StructField("product", StringType) :: Nil
      )

      // This is the output type of your aggregatation function.
      override def dataType: DataType = StringType

      override def deterministic: Boolean = true

      // This is the initial value for your buffer schema.
      override def initialize(buffer: MutableAggregationBuffer): Unit = {
        buffer(0) = ""
        buffer(1) = " "
      }

      // This is how to update your buffer schema given an input.
      override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
        buffer(0) = buffer.getAs[String](0) + 1
        buffer(1) = buffer.getAs[String](1) + input.getAs[String](0)
      }

      // This is how to merge two objects with the bufferSchema type.
      override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
        buffer1(0) = buffer1.getAs[Long](0) + buffer2.getAs[Long](0)
        buffer1(1) = buffer1.getAs[Double](1) * buffer2.getAs[Double](1)
      }

      // This is where you output the final value, given the final value of your bufferSchema.
      override def evaluate(buffer: Row): Any = {
        math.pow(buffer.getDouble(1), 1.toDouble / buffer.getLong(0))
      }
    }

}