Vectorized User-defined Functions

NOTE: Python UDF execution requires Python version (3.6, 3.7 or 3.8) with PyFlink installed. It’s required on both the client side and the cluster side.

Vectorized Python scalar functions take pandas.Series as the inputs and return a pandas.Series of the same length as the output. Internally, Flink will split the input elements into batches, convert a batch of input elements into and then call user-defined vectorized Python scalar functions for each batch of input elements. Please refer to the config option python.fn-execution.arrow.batch.size for more details on how to configure the batch size.

The following example shows how to define your own vectorized Python scalar function which computes the sum of two columns, and use it in a query:

Vectorized Aggregate Functions

Vectorized Python aggregate functions takes one or more pandas.Series as the inputs and return one scalar value as output.

Vectorized Python aggregate function could be used in GroupBy Aggregation(Batch), (Batch and Stream) and Over Window Aggregation(Batch and Stream bounded over window). For more details on the usage of Aggregations, you can refer to the relevant documentation.

Note Pandas UDAF does not support partial aggregation. Besides, all the data for a group or window will be loaded into memory at the same time during execution and so you must make sure that the data of a group or window could fit into the memory.

There are many ways to define a vectorized Python aggregate functions. The following examples show the different ways to define a vectorized Python aggregate function which takes two columns of bigint as the inputs and returns the sum of the maximum of them as the result.