问题标签 [pyflink]

有没有人使用 python 和 windows 和 flink 成功运行过？

我正在尝试以下命令：

.\bin\pyflink.bat examples\python\WordCount.py

并得到以下错误

我正在尝试 Flink 的新 Python 流 API 并尝试使用./flink-1.6.1/bin/pyflink-stream.sh examples/read_from_kafka.py. python 脚本相当简单，我只是尝试从现有主题中使用并将所有内容发送到标准输出（或输出方法默认发出数据的日志目录中的 *.out 文件）。

我从 maven repos 中抓取了一些 jar 文件，即flink-connector-kafka-0.9_2.11-1.6.1.jar，flink-connector-kafka-base_2.11-1.6.1.jar并将kafka-clients-0.9.0.1.jar它们复制到 Flink 的lib目录中。除非我误解了文档，否则这足以让 Flink 加载 kafka 连接器。事实上，如果我删除这些 jar 中的任何一个，导入就会失败，但这似乎不足以实际调用该计划。添加一个 for 循环来动态添加这些sys.path也不起作用。这是在控制台中打印的内容：

这是我在日志中看到的：

有没有办法解决这个问题并使连接器对 Python 可用？我怀疑这是 Jython 的 Classloader 问题，但我不知道如何进一步调查（也考虑到我不了解 Java）。非常感谢。

作为第一次尝试，我想从文件中读取 JSON 数据并将其传递给 Flink。我定义了一个源（逐行读取 JSON 字符串）和一个占位符过滤器。见代码：

当我构建作业并将其移动到我启动的 Flink 集群时，我收到以下错误消息：

PythonStreamExecutionEnvironment.execute(PythonStreamExecutionEnvironment.java:245) 在 sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) 在 sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) 在 sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java: 43) 在 java.lang.reflect.Method.invoke(Method.java:498) org.apache.flink.client.program.ProgramInvocationException: org.apache.flink.client.program.ProgramInvocationException: 作业失败。（职位编号：31615948194c951be03d46576929aa23）43) 在 java.lang.reflect.Method.invoke(Method.java:498) org.apache.flink.client.program.ProgramInvocationException: org.apache.flink.client.program.ProgramInvocationException: 作业失败。（职位编号：31615948194c951be03d46576929aa23）43) 在 java.lang.reflect.Method.invoke(Method.java:498) org.apache.flink.client.program.ProgramInvocationException: org.apache.flink.client.program.ProgramInvocationException: 作业失败。（职位编号：31615948194c951be03d46576929aa23）

该程序不包含 Flink 作业。也许您忘记在执行环境上调用 execute()。

我没有忘记调用execute()。

我想在 PyFlink 作业的 jar 文件中调用我的 Java 接口。官方文档中没有找到解决方案。

是否可以将 PyFlink 与 LightGBM 等 Python 机器学习库一起用于流应用程序？这有什么好的例子吗？

在 CentOS (CentOS 8.0.1905 (64bit)) 上，我尝试在 Flink(1.9) 上运行 Python(3.6.8) 版本的 WordCount 程序，如此处所述。我得到如下错误。相同的环境适用于 Java 版本的 WordCount 程序。我在这里想念什么？任何帮助是极大的赞赏。错误

5 月 1 日添加：
我在 apache-flink 的 jira 上看到了一个关于这个错误的问题，也许它有帮助？

我的系统是CentOS7，python版本3.6.8，pyflink版本1.10.0；
我正在关注本教程并尝试运行 pyflink 文件；但我不断收到以下错误；
我曾尝试使用选项 -pyarch 和 -pyexec venv.zip/venv/bin/python3运行它，但它没用。
我还将t_env.get_config().set_python_executable("python3")添加到 py 文件中，但仍然出现相同的错误：

我的默认 python 版本已设置为 Python3.6：

看来我应该设置一些环境变量？但我不知道应该设置什么名称和路径。现在我有一个 $FLINK_HOME。
任何帮助是极大的赞赏。

我是 Flink 的新手，我发现 Flink 支持 Python，称为 PyFlink。但我不知道在哪里可以找到相关的 PyFlink 文档或示例？

有没有办法专门使用 Python 对 flink 作业进行分析？据我了解，table api将检索数据，我们只能使用api中的函数来转换数据。我是否遗漏或误解了文档中的任何内容？

无论如何要在Java Flink作业中使用python用户定义的函数，或者无论如何都要传达例如由flink与java完成的转换结果与python用户定义的函数来应用一些机器学习的东西：

我知道从 pyFlink 你可以做这样的事情：

但是我需要做类似的事情，但是从 java 添加 python 函数，或者如何将 java 转换的结果直接传递给 Python UDF Flink 作业？

我希望这些问题不要太疯狂，但我需要知道是否存在以某种方式将 Flink DataStream API 与以 Java 作为主要语言的 Python Table API 进行通信？这意味着我需要从 Java 执行：Source -> Transformations -> Sink，但是其中一些转换可以触发 Python 函数，或者 Python 函数将等待一些 Java 转换完成以对 Stream 结果执行某些操作。

我希望有人明白我在这里想要做什么。

亲切的问候！