This topic describes how to create, register, and use a Python user-defined table-valued function (UDTF) in Realtime Compute for Apache Flink.
DefinitionA 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.
- 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
- Download and decompress the python_demo-master package to your on-premises machine.
- In the main menu bar of PyCharm, choose python_demo-master package. to open the decompressed
- 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, splits
- 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 Register a UDF.
Use a UDTF
- Use Flink SQL to create a job. For more information, see Develop a job.
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 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 (|)