Python

Types of user-defined functions

Category	Description
user-defined scalar function (UDSF)	A UDSF maps zero, one, or more scalar values to a new scalar value. It processes one input row to produce one output value, creating a one-to-one mapping. For more information, see User-defined scalar functions (UDSFs).
user-defined aggregate function (UDAF)	A UDAF aggregates multiple records into a single record, creating a many-to-one mapping. For more information, see User-defined aggregate functions (UDAFs).
user-defined table function (UDTF)	A UDTF accepts zero, one, or more scalar values as input parameters. Unlike a scalar function, a table function can return any number of rows, each consisting of one or more columns. For more information, see User-defined table functions (UDTFs).

Using Python dependencies

Realtime Compute for Apache Flink clusters come pre-installed with common Python packages such as Pandas, NumPy, and PyArrow. For a list of pre-installed third-party Python packages, see Develop Python jobs. Import these packages in your function before use, as shown in the following example.

@udf(result_type=DataTypes.FLOAT())
def percentile(values: List[float], percentile: float):
    import numpy as np
    return np.percentile(values, percentile)

To use a third-party Python package that is not pre-installed, upload it as a dependency file when you register the Python UDF. For more information, see Manage user-defined functions (UDFs) and Use Python dependencies.

Code debugging

You can use the logging module to output log information from your Python user-defined function for troubleshooting. The following example shows how to use logging.

@udf(result_type=DataTypes.BIGINT())
def add(i, j):    
  logging.info("hello world")    
  return i + j

After the logs are generated, you can view them in the TaskManager log files. For more information, see View run logs.

Performance tuning

Preloading resources

Preload resources during function initialization to avoid reloading them on each call to the eval method. For example, you can load a large deep learning model once and then run batch predictions against it.

from pyflink.table import DataTypes
from pyflink.table.udf import ScalarFunction, udf

class Predict(ScalarFunction):
    def open(self, function_context):
        import pickle

        with open("resources.zip/resources/model.pkl", "rb") as f:
            self.model = pickle.load(f)

    def eval(self, x):
        return self.model.predict(x)

predict = udf(Predict(), result_type=DataTypes.DOUBLE(), func_type="pandas")

Note

For information about how to upload Python data files, see Use Python dependencies.

Asynchronous user-defined functions

For I/O-intensive scenarios, such as accessing external databases or making HTTP service calls, use asynchronous user-defined functions. A single function instance can handle multiple requests concurrently, distributing the wait time across several calls and significantly improving job throughput. This feature is supported only in VVR 11.7 and later versions and applies only to user-defined scalar functions (UDSFs). For details, see Asynchronous user-defined functions.

Using the Pandas library

In addition to standard Python user-defined functions, Realtime Compute for Apache Flink supports Pandas user-defined functions. These functions accept Pandas data structures such as pandas.Series and pandas.DataFrame as input, allowing you to leverage high-performance libraries like Pandas and NumPy. For more information, see Vectorized User-defined Functions.

Parameters

The performance of a Python user-defined function largely depends on its implementation. If you encounter performance issues, optimize the function logic first. The following parameters also affect performance.

Parameter	Description
python.fn-execution.bundle.size	Python UDFs execute asynchronously. The Java operator caches data before sending it to a Python process for execution. When the cache reaches a threshold, the data is sent to the Python process. The `python.fn-execution.bundle.size` parameter controls the maximum number of records that can be cached. The default value is 100,000 records.
python.fn-execution.bundle.time	This parameter controls the maximum caching time. The cached data is sent for processing when either the record count reaches the `python.fn-execution.bundle.size` threshold or the caching time reaches the `python.fn-execution.bundle.time` threshold. The default value is 1,000 milliseconds.
python.fn-execution.arrow.batch.size	For Pandas UDFs, this parameter specifies the maximum number of records an Arrow batch can contain. The default value is 10,000. Note The value of the `python.fn-execution.arrow.batch.size` parameter cannot be greater than the value of the `python.fn-execution.bundle.size` parameter.