实时计算Flink版：自定义表值函数（UDTF）

定义

自定义表值函数（UDTF），将0个、1个或多个标量值作为输入参数（可以是变长参数）。表值函数可以返回任意数量的行作为输出，而不仅是1个值。返回的行可以由1个或多个列组成。调用一次函数输出多行或多列数据。与自定义的标量函数类似，但与标量函数不同。

使用限制

由于实时计算Flink版受部署环境和网络环境等因素的影响，开发Python自定义函数时，需要注意以下限制：

• 仅支持开源Flink V1.12及以上版本。

• Flink工作空间已预装了Python，因此需要您在对应Python版本上开发代码。

说明

实时计算引擎VVR 8.0.11以下版本预装Python 3.7.9版本，实时计算引擎VVR 8.0.11及以上版本预装Python 3.9.21版本。如需将低版本升级至实时计算引擎VVR 8.0.11及以上版本，必须对先前版本的PyFlink作业进行重新测试、部署和运行。

• Flink运行环境仅支持JDK 8和JDK 11，如果Python作业中依赖第三方JAR包，请确保JAR包兼容。

• 仅支持开源Scala V2.11版本，如果Python作业中依赖第三方JAR包，请确保使用Scala V2.11对应的JAR包依赖。

• 内联函数仅在VVR 11.9.0 及以上版本中支持。

UDTF开发

用户可以通过以下两种方式开发对应的Python函数：

• 将Python代码打包、上传并注册进入平台的方式；

• 在SQL脚本中通过内联函数形式声明代码逻辑；

如果UDTF计算逻辑简单，您可以尝试使用内联函数的形式来开发Python函数。

基于代码打包的UDTF开发

    from pyflink.table import DataTypes
    from pyflink.table.udf import udtf
    
    @udtf(result_types=[DataTypes.STRING(), DataTypes.STRING()])
    def my_split(s: str):
        splits = s.split("|")
        yield splits[0], splits[1]
        

    zip -r python_demo.zip udx
        

    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(my_split(concat_ws('|', `message`, 'aa'))) as T(name,place);
        

Python内联表函数

内联函数指函数实现直接嵌入在CREATE FUNCTION语句中，函数定义与注册声明在同一条 SQL 语句内完成。例如，以下就是一个对于邮箱地址进行脱敏的内联函数，开发者需要将完整的代码逻辑声明在$$之间。

CREATE TEMPORARY FUNCTION parse_order_items(items_json STRING)
RETURNS TABLE (
    sku STRING,
    quantity INT
)
HANDLER 'ParseOrderItems'
AS $$
import json

class ParseOrderItems:
    def eval(self, items_json):
        if not items_json:
            return

        try:
            items = json.loads(items_json)
        except (TypeError, ValueError):
            return

        if not isinstance(items, list):
            return

        for item in items:
            if not isinstance(item, dict):
                continue

            sku = str(item.get("sku", "")).strip().upper()

            try:
                quantity = int(item.get("quantity"))
            except (TypeError, ValueError):
                continue

            if sku and quantity > 0:
                yield sku, quantity
$$
LANGUAGE PYTHON;