1
我想编写Spark UDAF,其中列的类型可以是任何具有定义在其上的Scala数字的列。我在互联网上搜索过,但只找到像DoubleType,LongType这样的具体类型的例子。这不可能吗?但是,如何将UDAF与其他数值一起使用呢?Spark UDAF - 使用泛型作为输入类型?
我想编写Spark UDAF,其中列的类型可以是任何具有定义在其上的Scala数字的列。我在互联网上搜索过,但只找到像DoubleType,LongType这样的具体类型的例子。这不可能吗?但是,如何将UDAF与其他数值一起使用呢?Spark UDAF - 使用泛型作为输入类型?
为简单起见,我们假设您想定义一个自定义sum
。你必须提供输入类型TypeTag
并使用Scala的反射来定义模式:
import org.apache.spark.sql.expressions._
import org.apache.spark.sql.types._
import org.apache.spark.sql.Row
import scala.reflect.runtime.universe._
import org.apache.spark.sql.catalyst.ScalaReflection.schemaFor
case class MySum [T : TypeTag](implicit n: Numeric[T])
extends UserDefinedAggregateFunction {
val dt = schemaFor[T].dataType
def inputSchema = new StructType().add("x", dt)
def bufferSchema = new StructType().add("x", dt)
def dataType = dt
def deterministic = true
def initialize(buffer: MutableAggregationBuffer) = buffer.update(0, n.zero)
def update(buffer: MutableAggregationBuffer, input: Row) = {
if (!input.isNullAt(0))
buffer.update(0, n.plus(buffer.getAs[T](0), input.getAs[T](0)))
}
def merge(buffer1: MutableAggregationBuffer, buffer2: Row) = {
buffer1.update(0, n.plus(buffer1.getAs[T](0), buffer2.getAs[T](0)))
}
def evaluate(buffer: Row) = buffer.getAs[T](0)
}
定义为上面我们可以创建实例处理特定类型的函数:
val sumOfLong = MySum[Long]
spark.range(10).select(sumOfLong($"id")).show
+---------+
|mysum(id)|
+---------+
| 45|
+---------+
注意:
为了获得与内置集合函数相同的灵活性,您必须定义您自己的AggregateFunction
,如ImperativeAggregate
或DeclarativeAggregate
。这是可能的,但它是一个内部API。