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:

Development methods

You can develop a Python UDSF in either of the following ways:

• Package your Python code, upload the package, and register it as a function on the platform.

• Declare the code logic as an inline function in a SQL statement.

If the logic of your UDSF is simple, consider developing it as an inline function.

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.

Python inline scalar functions

An inline function embeds its implementation directly in a CREATE FUNCTION statement, so the function is defined and registered within the same SQL statement. The following example defines an inline function that masks email addresses. Declare the complete code logic between the $$ delimiters.

CREATE TEMPORARY FUNCTION mask_email(email STRING)
RETURNS STRING AS $$
if email is None:
    return None

name, separator, domain = email.partition("@")
if not separator:
    return email

return name[:1] + "***@" + domain
$$ LANGUAGE PYTHON;

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()
)