Timo,
I converted what I had to Java, and ended up with the exact same issue as
before where it will work if I only ever use it on 1 type, but not if I use it
on multiple. Maybe this is a bug?
Dylan
On 1/20/21, 10:06 AM, "Dylan Forciea" <dy...@oseberg.io> wrote:
Oh, I think I might have a clue as to what is going on. I notice that it
will work properly when I only call it on Long. I think that it is using the
same generated code for the Converter for whatever was called first.
Since in Scala I can't declare an object as static within the class
itself, I wonder if it won't generate appropriate Converter code per subtype. I
tried creating a subclass that is specific to the type within my class and
returning that as the accumulator, but that didn't help. And, I can't refer to
that class in the TypeInference since it isn't static and I get an error from
Flink because of that. I'm going to see if I just write this UDF in Java with
an embedded public static class like you have if it will solve my problems.
I'll report back to let you know what I find. If that works, I'm not quite sure
how to make it work in Scala.
Regards,
Dylan Forciea
On 1/20/21, 9:34 AM, "Dylan Forciea" <dy...@oseberg.io> wrote:
As a side note, I also just tried to unify into a single function
registration and used _ as the type parameter in the classOf calls there and
within the TypeInference definition for the accumulator and still ended up with
the exact same stack trace.
Dylan
On 1/20/21, 9:22 AM, "Dylan Forciea" <dy...@oseberg.io> wrote:
Timo,
I appreciate it! I am using Flink 1.12.0 right now with the Blink
planner. What you proposed is roughly what I had come up with the first time
around that resulted in the stack trace with the ClassCastException I had
originally included. I saw that you had used a Row instead of just the value in
our example, but changing it that way didn't seem to help, which makes sense
since the problem seems to be in the code generated for the accumulator
Converter and not the output.
Here is the exact code that caused that error (while calling
LatestNonNullLong):
The registration of the below:
env.createTemporarySystemFunction("LatestNonNullLong",
classOf[LatestNonNull[Long]])
env.createTemporarySystemFunction("LatestNonNullString",
classOf[LatestNonNull[String]])
The class itself:
import java.time.LocalDate
import java.util.Optional
import org.apache.flink.table.api.DataTypes
import org.apache.flink.table.catalog.DataTypeFactory
import org.apache.flink.table.functions.AggregateFunction
import
org.apache.flink.table.types.inference.{InputTypeStrategies, TypeInference}
case class LatestNonNullAccumulator[T](
var value: T = null.asInstanceOf[T],
var date: LocalDate = null)
class LatestNonNull[T] extends AggregateFunction[T,
LatestNonNullAccumulator[T]] {
override def createAccumulator(): LatestNonNullAccumulator[T] = {
LatestNonNullAccumulator[T]()
}
override def getValue(acc: LatestNonNullAccumulator[T]): T = {
acc.value
}
def accumulate(acc: LatestNonNullAccumulator[T], value: T, date:
LocalDate): Unit = {
if (value != null) {
Option(acc.date).fold {
acc.value = value
acc.date = date
} { accDate =>
if (date != null && date.isAfter(accDate)) {
acc.value = value
acc.date = date
}
}
}
}
def merge(
acc: LatestNonNullAccumulator[T],
it: java.lang.Iterable[LatestNonNullAccumulator[T]]): Unit =
{
val iter = it.iterator()
while (iter.hasNext) {
val a = iter.next()
if (a.value != null) {
Option(acc.date).fold {
acc.value = a.value
acc.date = a.date
} { accDate =>
Option(a.date).map { curDate =>
if (curDate.isAfter(accDate)) {
acc.value = a.value
acc.date = a.date
}
}
}
}
}
}
def resetAccumulator(acc: LatestNonNullAccumulator[T]): Unit = {
acc.value = null.asInstanceOf[T]
acc.date = null
}
override def getTypeInference(typeFactory: DataTypeFactory):
TypeInference = {
TypeInference
.newBuilder()
.inputTypeStrategy(InputTypeStrategies
.sequence(InputTypeStrategies.ANY,
InputTypeStrategies.explicit(DataTypes.DATE())))
.accumulatorTypeStrategy { callContext =>
val accDataType = DataTypes.STRUCTURED(
classOf[LatestNonNullAccumulator[T]],
DataTypes.FIELD("value",
callContext.getArgumentDataTypes.get(0)),
DataTypes.FIELD("date", DataTypes.DATE()))
Optional.of(accDataType)
}
.outputTypeStrategy { callContext =>
val outputDataType =
callContext.getArgumentDataTypes().get(0);
Optional.of(outputDataType);
}
.build()
}
}
Regards,
Dylan Forciea
On 1/20/21, 2:37 AM, "Timo Walther" <twal...@apache.org> wrote:
Hi Dylan,
I'm assuming your are using Flink 1.12 and the Blink planner?
Beginning from 1.12 you can use the "new" aggregate functions
with a
better type inference. So TypeInformation will not be used in
this stack.
I tried to come up with an example that should explain the
rough design.
I will include this example into the Flink code base. I hope
this helps:
import
org.apache.flink.table.types.inference.InputTypeStrategies;
public static class LastIfNotNull<T>
extends AggregateFunction<Row,
LastIfNotNull.Accumulator<T>> {
public static class Accumulator<T> {
public T value;
public LocalDate date;
}
public void accumulate(Accumulator<T> acc, T input,
LocalDate date) {
if (input != null) {
acc.value = input;
acc.date = date;
}
}
@Override
public Row getValue(Accumulator<T> acc) {
return Row.of(acc.value, acc.date);
}
@Override
public Accumulator<T> createAccumulator() {
return new Accumulator<>();
}
@Override
public TypeInference getTypeInference(DataTypeFactory
typeFactory) {
return TypeInference.newBuilder()
.inputTypeStrategy(
InputTypeStrategies.sequence(
InputTypeStrategies.ANY,
InputTypeStrategies.explicit(DataTypes.DATE())))
.accumulatorTypeStrategy(
callContext -> {
DataType accDataType =
DataTypes.STRUCTURED(
Accumulator.class,
DataTypes.FIELD(
"value",
callContext.getArgumentDataTypes().get(0)),
DataTypes.FIELD("date",
DataTypes.DATE()));
return Optional.of(accDataType);
})
.outputTypeStrategy(
callContext -> {
DataType argDataType =
callContext.getArgumentDataTypes().get(0);
DataType outputDataType =
DataTypes.ROW(
DataTypes.FIELD("value",
argDataType),
DataTypes.FIELD("date",
DataTypes.DATE()));
return
Optional.of(outputDataType);
})
.build();
}
}
Regards,
Timo
On 20.01.21 01:04, Dylan Forciea wrote:
> I am attempting to create an aggregate UDF that takes a
generic
> parameter T, but for the life of me, I can’t seem to get it
to work.
>
> The UDF I’m trying to implement takes two input arguments, a
value that
> is generic, and a date. It will choose the non-null value
with the
> latest associated date. I had originally done this with
separate Top 1
> queries connected with a left join, but the memory usage
seems far
> higher than doing this with a custom aggregate function.
>
> As a first attempt, I tried to use custom type inference to
have it
> validate that the first argument type is the output type and
have a
> single function, and also used DataTypes.STRUCTURE to try to
define the
> shape of my accumulator. However, that resulted in an
exception like
> this whenever I tried to use a non-string value as the first
argument:
>
> [error] Caused by: java.lang.ClassCastException:
java.lang.Long cannot
> be cast to java.lang.String
>
> [error] at
>
io$oseberg$flink$udf$LatestNonNullAccumulator$Converter.toInternal(Unknown
> Source)
>
> [error] at
>
org.apache.flink.table.data.conversion.StructuredObjectConverter.toInternal(StructuredObjectConverter.java:92)
>
> [error] at
>
org.apache.flink.table.data.conversion.StructuredObjectConverter.toInternal(StructuredObjectConverter.java:47)
>
> [error] at
>
org.apache.flink.table.data.conversion.DataStructureConverter.toInternalOrNull(DataStructureConverter.java:59)
>
> [error] at GroupAggsHandler$777.getAccumulators(Unknown
Source)
>
> [error] at
>
org.apache.flink.table.runtime.operators.aggregate.GroupAggFunction.processElement(GroupAggFunction.java:175)
>
> [error] at
>
org.apache.flink.table.runtime.operators.aggregate.GroupAggFunction.processElement(GroupAggFunction.java:45)
>
> [error] at
>
org.apache.flink.streaming.api.operators.KeyedProcessOperator.processElement(KeyedProcessOperator.java:85)
>
> [error] at
>
org.apache.flink.streaming.runtime.tasks.OneInputStreamTask$StreamTaskNetworkOutput.emitRecord(OneInputStreamTask.java:193)
>
> [error] at
>
org.apache.flink.streaming.runtime.io.StreamTaskNetworkInput.processElement(StreamTaskNetworkInput.java:179)
>
> [error] at
>
org.apache.flink.streaming.runtime.io.StreamTaskNetworkInput.emitNext(StreamTaskNetworkInput.java:152)
>
> [error] at
>
org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:67)
>
> [error] at
>
org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:372)
>
> [error] at
>
org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:186)
>
> [error] at
>
org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:575)
>
> [error] at
>
org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:539)
>
> [error] at
org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:722)
>
> [error] at
org.apache.flink.runtime.taskmanager.Task.run(Task.java:547)
>
> [error] at java.lang.Thread.run(Thread.java:748)
>
> Figuring that I can’t do something of that sort, I tried to
follow the
> general approach in the Sum accumulator[1] in the Flink
source code
> where separate classes are derived from a base class, and
each
> advertises its accumulator shape, but ended up with the
exact same stack
> trace as above when I tried to create and use a function
specifically
> for a non-string type like Long.
>
> Is there something I’m missing as far as how this is
supposed to be
> done? Everything I try either results in a stack track like
the above,
> or type erasure issues when trying to get type information
for the
> accumulator. If I just copy the generic code multiple times
and just
> directly use Long or String rather than using subclassing,
then it works
> just fine. I appreciate any help I can get on this!
>
> Regards,
>
> Dylan Forciea
>
> [1]
>
https://github.com/apache/flink/blob/release-1.12.0/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/functions/aggfunctions/SumAggFunction.scala
>
