Apache Flink 1.14.0 Java通过SQL DDL调用Python UDF失败求助

问题原因 #

你遇到的报错核心是Java作业运行环境缺少Python UDF加载必要的依赖和配置，SQL Client默认内置了PyFlink相关组件所以可以正常运行，自定义Java项目需要额外适配：

1. 项目未引入flink-python依赖，Flink无法识别Python UDF的注册、实例化逻辑
2. 仅配置了客户端侧Python执行路径，未配置TaskManager运行时的Python执行路径
3. 存在重复执行逻辑冲突：executeSql执行查询语句时已触发作业提交，后续env.execute()属于多余调用

修复步骤 #

1. 添加flink-python依赖 #

在你的Java项目pom.xml中引入和集群版本匹配的flink-python依赖：

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-python_2.12</artifactId>
    <version>1.14.0</version>
    <!-- 若集群lib目录已放置该jar包则scope为provided，否则需要打入fat jar -->
    <scope>provided</scope>
</dependency>

2. 补全Python配置 #

在原有配置基础上新增TaskManager侧Python执行器配置：

// 原有配置保留
tEnv.getConfig().getConfiguration().setString("python.files", "/home/magic/workspace/python/flinkTestUdf/udfTest.py");
tEnv.getConfig().getConfiguration().setString("python.client.executable", "python3");
// 新增运行时Python执行路径配置
tEnv.getConfig().getConfiguration().setString("python.executable", "python3");

3. 删除多余执行逻辑 #

删除代码末尾的env.execute();语句，TableResult.print()已经会触发作业执行并拉取结果。

可选校验点 #

如果修改后仍报错，可以做如下检查：

• 确保所有TaskManager节点的/home/magic/workspace/python/flinkTestUdf/udfTest.py路径存在对应文件
• 把CREATE TEMPORARY SYSTEM FUNCTION中的SYSTEM关键字去掉，使用普通临时函数注册

内容的提问来源于stack exchange，提问作者Liam Zee