pyflink自定义udf

在 PyFlink 中自定义 UDF（User-Defined Function），可以使用 Python 编写 UDF 函数，并将其注册到 Flink 中。下面是一个简单的例子，演示如何使用 Python 编写 UDF 函数，并在 PyFlink 中注册和使用它：

from pyflink.table.udf import udf
from pyflink.table import ScalarFunction
from pyflink.table import DataTypes
from pyflink.table import EnvironmentSettings

# 自定义 UDF 函数
@udf(result_type=DataTypes.STRING(), func_type="scalar")
def my_upper(s: str) -> str:
    return s.upper()

# 创建 PyFlink 表环境
env_settings = EnvironmentSettings.new_instance().in_batch_mode().use_blink_planner().build()
t_env = BatchTableEnvironment.create(environment_settings=env_settings)

# 注册 UDF 函数
t_env.register_function("my_upper", my_upper)

# 使用 UDF 函数
input_path = "/path/to/input"
output_path = "/path/to/output"
t_env.execute_sql(f"""
    CREATE TABLE my_input (
        id INT,
        name STRING
    ) WITH (
        'connector' = 'csv',
        'format' = 'csv',
        'path' = '{input_path}'
    )
""")
t_env.execute_sql(f"""
    CREATE TABLE my_output (
        id INT,
        name_upper STRING
    ) WITH (
        'connector' = 'csv',
        'format' = 'csv',
        'path' = '{output_path}'
    )
""")
t_env.execute_sql("""
    INSERT INTO my_output
    SELECT id, my_upper(name) AS name_upper
    FROM my_input
""")

在上面的例子中，我们使用 `@udf` 装饰器定义了一个名为 `my_upper` 的 UDF 函数，用于将字符串转换为大写。然后，我们在 PyFlink 中注册了这个 UDF 函数，使用 `t_env.register_function` 方法将 `my_upper` 函数注册到 Flink 中。最后，我们使用 `my_upper` 函数将输入表中的字符串转换为大写，并将结果写入输出表中。