Maybe instead of trying to implement abbreviated UDAF on tumbling window using streaming mode (i.e. Kafka/kinesis):

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

I need to be more verbose, and define some kind of Class [1] that extends `AggregateFunction`:

class WeightedAvg(AggregateFunction):

    def create_accumulator(self):
        # Row(sum, count)
        return Row(0, 0)

    def get_value(self, accumulator):
        if accumulator[1] == 0:
            return None
        else:
            return accumulator[0] / accumulator[1]

    def accumulate(self, accumulator, value, weight):
        accumulator[0] += value * weight
        accumulator[1] += weight
    
    def retract(self, accumulator, value, weight):
        accumulator[0] -= value * weight
        accumulator[1] -= weight
        
    def get_result_type(self):
        return DataTypes.BIGINT()
        
    def get_accumulator_type(self):
        return DataTypes.ROW([
            DataTypes.FIELD("f0", DataTypes.BIGINT()), 
            DataTypes.FIELD("f1", DataTypes.BIGINT())])

Can I convert the data structure to pandas maybe [2], either before implementing the tumbling-window/UDAF, or within the UDAF? That way I can implement pandas notation somehow in the above

AggregateFunction

extension, rather than trying to do UDAF in tumbling-window stream mode which yields the error:

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

--- [1] https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/python/table/udfs/python_udfs/#aggregate-functions [2] https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/python/table/conversion_of_pandas/