Apache Flink 1.11在Java Flink Streaming Job中无法通过SQL函数DDL使用Pyt

Question

在Flip-106中，有一个如何通过SQL函数DDL在批处理作业java应用程序中调用用户定义的python函数的示例。。。

BatchTableEnvironment tEnv = BatchTableEnvironment.create(env);
tEnv.getConfig().getConfiguration().setString("python.files", "/home/my/test1.py");
tEnv.getConfig().getConfiguration().setString("python.client.executable", "python3");
tEnv.sqlUpdate("create temporary system function func1 as 'test1.func1' language python");
Table table = tEnv.fromDataSet(env.fromElements("1", "2", "3")).as("str").select("func1(str)");
tEnv.toDataSet(table, String.class).collect();

我一直试图在流式作业java应用程序中重现这个相同的例子，这是我的代码：

final StreamTableEnvironment fsTableEnv = StreamTableEnvironment.create(EnvironmentConfiguration.getEnv(), fsSettings);
fsTableEnv.getConfig().getConfiguration().setString("python.files", "/Users/jf/Desktop/flink/fca/test.py");
fsTableEnv.getConfig().getConfiguration().setString("python.client.executable", "/Users/jf/opt/anaconda3/bin/python");
fsTableEnv.sqlUpdate("CREATE TEMPORARY SYSTEM FUNCTION func1 AS 'test.func1' LANGUAGE PYTHON");
Table table = fsTableEnv.fromValues("1", "2", "3").as("str").select("func1(str)");
/* Missing line */

对于批处理作业中的这一特定行：

tEnv.toDataSet(table, String.class).collect();

我还没有找到流媒体工作的对等

1.你能帮我把这个flip-106的例子从批处理映射到流吗

我最终想要的是用flink 1.11调用流化作业中的python函数java flink应用程序，如下所示：

final StreamTableEnvironment fsTableEnv = StreamTableEnvironment.create(EnvironmentConfiguration.getEnv(), fsSettings);
fsTableEnv.getConfig().getConfiguration().setString("python.files", "/Users/jf/Desktop/flink/fca/test.py");
fsTableEnv.getConfig().getConfiguration().setString("python.client.executable", "/Users/jf/opt/anaconda3/bin/python");
fsTableEnv.sqlUpdate("CREATE TEMPORARY SYSTEM FUNCTION func1 AS 'test.func1' LANGUAGE PYTHON");
final Table table = fsTableEnv.fromDataStream(stream_filtered.map(x->x.idsUmid)).select("func1(f0)").as("umid");
System.out.println("Result --> " + table.select($("umid")) + " --> End of Result");

并使用udf的结果进行进一步处理(不一定在控制台中打印(

我编辑了test.py文件，以查看是否至少不管未命名的表，都在python中执行了某些操作。

from pyflink.table.types import DataTypes
from pyflink.table.udf import udf
from os import getcwd
@udf(input_types=[DataTypes.STRING()], result_type=DataTypes.STRING())
def func1(line):
print(line)
print(getcwd())
with open("test.txt", "a") as myfile:
myfile.write(line)
return line

并且不打印任何内容，也不创建test.txt文件，也不将值返回到流作业。所以基本上这个python函数没有被调用。

2.我这里缺少什么

感谢David、Wei和Xingbo到目前为止的支持，因为建议的每一个细节都对我有效。

谨致问候，

Jonathan

Answer 1

你可以试试这个：

final StreamTableEnvironment fsTableEnv = StreamTableEnvironment.create(EnvironmentConfiguration.getEnv(), fsSettings);
fsTableEnv.getConfig().getConfiguration().setString("python.files", "/Users/jf/Desktop/flink/fca/test.py");
fsTableEnv.getConfig().getConfiguration().setString("python.client.executable", "/Users/jf/opt/anaconda3/bin/python");
// You need to specify the python interpreter used to run the python udf on cluster.
// I assume this is a local program so it is the same as the "python.client.executable".
fsTableEnv.getConfig().getConfiguration().setString("python.executable", "/Users/jf/opt/anaconda3/bin/python");
fsTableEnv.sqlUpdate("CREATE TEMPORARY SYSTEM FUNCTION func1 AS 'test.func1' LANGUAGE PYTHON");
final Table table = fsTableEnv.fromDataStream(stream_filtered.map(x->x.idsUmid)).select("func1(f0)").as("umid");
// 'table.select($("umid"))' will not trigger job execution. You need to call the "execute()" method explicitly.
table.execute().print();