UDTFs - Realtime Compute for Apache Flink - Alibaba Cloud Documentation Center

This topic describes how to create, register, and use a Python user-defined table-valued function (UDTF) in Realtime Compute for Apache Flink.

Definition

A UDTF takes zero, one, or more scalar values as input parameters. These parameters can be variable-length parameters. UDTFs return any number of rows instead of a single value. Returned rows consist of one or more columns. Multiple rows or columns are returned each time a UDTF is called. UDTFs are similar to user-defined scalar functions (UDFs) but differ in the results they return.

Limits

Services provided by fully managed Flink are subject to deployment environments and network environments. Therefore, when you develop Python UDFs in fully managed Flink, take note of the following limits:

Only Apache Flink 1.12 and later are supported.
Python 3.7.9 is pre-installed on a fully managed Flink cluster. Therefore, you must develop code in Python 3.7.9.
JDK 1.8 is used in the runtime environment of fully managed Flink. If your Python job depends on a third-party JAR package, make sure that the JAR package is compatible with JDK 1.8.
Only open source Scala 2.11 is supported. If your Python job depends on a third-party JAR package, make sure that the JAR package that corresponds to Scala 2.11 is used.

Create a UDTF

Note

Flink provides sample code of Python user-defined extensions (UDXs) for you to develop UDXs. The sample code includes the implementation of Python UDFs, Python user-defined aggregate functions (UDAFs), and Python UDTFs. This topic describes how to create a UDTF in the Windows operating system.

Download and decompress the python_demo-master package to your on-premises machine.
In the main menu bar of PyCharm, choose File > Open to open the decompressed python_demo-master package.
Double-click the udtfs.py file in the \python_demo-master\udx directory. Then, modify the content of the file based on your business requirements.
In this example, split defines the code that can separate a row of string into multiple columns of strings with vertical bars (|).
```
from pyflink.table import DataTypes
from pyflink.table.udf import udtf

@udtf(result_types=[DataTypes.STRING(), DataTypes.STRING()])
def split(s: str):
    splits = s.split("|")
    yield splits[0], splits[1]
```
Go to the \python_demo directory to which the udx folder belongs and run the following command to package the files in the directory:
```
zip -r python_demo.zip udx
```
If the python_demo.zip package appears in the \python_demo\ directory, the UDTF is developed.

Register a UDTF

For more information about how to register a UDTF, see Manage UDFs.

Use a UDTF

After you register a UDTF, you can perform the following steps to use the UDTF:

Use Flink SQL to create a job. For more information, see Develop an SQL draft.

After the aa string and the message field in each row of string in the ASI_UDTF_Source table are concatenated with vertical bars (|), the concatenated strings are separated into multiple columns of strings by vertical bars (|). The following code shows an example:

CREATE TEMPORARY TABLE ASI_UDTF_Source (
  `message`  VARCHAR
) WITH (
  'connector'='datagen'
);

CREATE TEMPORARY TABLE ASI_UDTF_Sink (
  name  VARCHAR,
  place  VARCHAR
) WITH (
  'connector' = 'blackhole'
);

INSERT INTO ASI_UDTF_Sink
SELECT name,place
FROM ASI_UDTF_Source,lateral table(split(concat_ws('|', `message`, 'aa'))) as T(name,place);

On the O&M > Deployments page in the console of fully managed Flink, find the job that you want to start, and click Start in the Actions column.
After the job is started, two columns of data are inserted into the ASI_UDTF_Sink table. The two columns of data contain the concatenated strings that are separated by vertical bars (|)