I need a median function, and some quantiles. I wrote this

@udaf(result_type=DataTypes.FLOAT(), func_type="pandas")
def median_udaf(v):
    return v.median()

unfortunately it doesn't work due to

Pandas UDAFs are not supported in streaming mode currently.
Please check the documentation for the set of currently supported SQL features.

What can I do? We currently use 1.16, but I figure 1.17 is just around the corner so that could be used as well.

Are you computing the median value on an unbounded stream?

@Dian Fu No, it’s grouped by to one hour tumbling windows with SQL syntax. It occurred to me that it might be an SQL syntax limitation. Maybe Python table API would work?

I think there are some problems with your code. Could you share the code?

I'm not sure, my employer is very security sensitive.

OK. Then you could double check the code. From the exception stack, it’s not computing aggregation on a window, instead on a unbounded stream.

I'll check. I also talked internally, and we think this solution would be too slow. Ideally I would like to have a way to call this t-digest algorithm coded in Java, but with a way to call it from pyflink code. https://github.com/tdunning/t-digest Maybe you would have suggestions on how to achieve that?

Do you mean to call

t-digest

in Python UDAF?

I intend to write a version that calls the aggregate function that uses these functions, and not the vectorized version: •

create_accumulator()

•

accumulate(...)

•

get_value(...)

The python functions will then call Java (t-digest) using Py4J (pyflink.java_gateway).

OK，got it. I guess it should work.

@Dian Fu It turned out that it doesn't work. The accumulator needs to be a Flink datatype, and it would need to be a Java class for t-digest to be usable. There's code that checks it, and it returns an error. The

retract(...)

API call also is not possible with t-digest as you cannot remove data from it as it's already been combined. In my opinion the API is not particularly good, but it is what it is.

@Ari Huttunen OK, thanks for updating the status. It seems that you are also using Java object to represent the immediate result (accumulator). Are you performing unbounded aggregation or window aggregation? For the latter case, things may become simpler as you don’t need to define the accumulator and also no retract is needed.