Realtime Compute for Apache Flink:Python user-defined scalar functions (UDSFs)

Limits

The following constraints apply when you develop Python user-defined functions (UDFs) in Realtime Compute for Apache Flink:

Create a UDSF

The following steps use Windows as the example environment. Flink provides a sample repository that includes implementations for UDSFs, user-defined aggregate functions (UDAFs), and user-defined table-valued functions (UDTFs).

1. Download and decompress python_demo-master to your local machine.

   This is a third-party GitHub repository. Access may be slow or intermittent.

2. In PyCharm, choose File > Open and open the decompressed python_demo-master directory.

3. Open udfs.py in the \python_demo-master\udx path and define your UDSF.

   from pyflink.table import DataTypes
   from pyflink.table.udf import udf
   
   @udf(result_type=DataTypes.STRING())
   def sub_string(s: str, begin: int, end: int):
       return s[begin:end]

   The sub_string example extracts characters from position begin to position end in the input string.

4. From the \python_demo-master directory, run the following command to package the udx directory:

   zip -r python_demo.zip udx

   When python_demo.zip appears in \python_demo-master\, the package is ready.

Register a UDSF

After creating the package, register the UDSF in the Realtime Compute for Apache Flink console. For registration steps, see Manage user-defined functions (UDFs).

Use a UDSF

After registering the UDSF, use it in a Flink SQL job.

1. Create a draft using Flink SQL. For details, see Job development overview. The following example calls ASI_UDSF (the registered name of your UDSF) to extract characters from positions 2 to 4 of the a field in the source table:

   CREATE TEMPORARY TABLE ASI_UDSF_Source (
     a VARCHAR,
     b INT,
     c INT
   ) WITH (
     'connector' = 'datagen'
   );
   
   CREATE TEMPORARY TABLE ASI_UDSF_Sink (
     a VARCHAR
   ) WITH (
     'connector' = 'blackhole'
   );
   
   INSERT INTO ASI_UDSF_Sink
   SELECT ASI_UDSF(a, 2, 4)
   FROM ASI_UDSF_Source;

2. In the left-side navigation pane of the development console, choose O&M > Deployments. Find the deployment, then click Start in the Actions column. After the deployment starts, characters at positions 2–4 of the a field in ASI_UDSF_Source are written to ASI_UDSF_Sink.

Asynchronous user-defined functions

For UDFs that perform I/O-intensive operations such as external database access or HTTP requests, use asynchronous user-defined functions. A single asynchronous UDF can handle multiple I/O requests concurrently, distributing wait time across requests and improving job throughput.

import asyncio

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


# Method 1: Use a Python async function
@udf(result_type=DataTypes.STRING())
async def async_api_call(product_id: str) -> str:
    await asyncio.sleep(0.05)
    return f"product_{product_id}"


# Method 2: Subclass the asynchronous function class
class AsyncUserLookup(AsyncScalarFunction):
    def open(self, function_context):
        self.cache = {}

    async def eval(self, user_id: str) -> str:
        if user_id in self.cache:
            return self.cache[user_id]

        await asyncio.sleep(0.05)
        result = f"user_{user_id}"
        self.cache[user_id] = result
        return result

    def close(self):
        self.cache.clear()


async_user_lookup = udf(
    AsyncUserLookup(),
    input_types=[DataTypes.STRING()],
    result_type=DataTypes.STRING()
)