使用pyflink从本地系统以批处理模式读取CSV文件
我在编写pyflink作业时尝试读取已建立的CSV文件。我使用文件系统连接器来获取数据，但是在DDL上执行EXECUTE_SQL()并稍后在表上执行查询之后，我收到一个错误，说明它无法获取下一个结果。我无法解决此错误。我已经检查了CSV文件，它是完全正确的，并且可以与 pandas 一起工作，但是在这里我不明白为什么它不能获取下一行。请查找附加的代码以供参考。 from pyflink.co
..
pyflink相关内容
pyflink tableAPI，多个源到单个处理表序列
我正在尝试实现一个pyflink作业(通过Table API)，在来自多个源的数据转换为标准格式后，该作业从多个源执行一些基本处理。我可以将每个源中的数据转换为所需的格式(具有指定列的“表架构”)，但不能将该数据设置为“处理表”以供进一步实现。这就是我要做的： 若要将不同的源架构转换为标准的处理表方案，然后可以对其进行窗口化，请实现表函数(针对多个输入行)等。 编辑： 感谢您建议
..
Apache Flink:Python 流式 API 中的 Kafka 连接器，“无法加载用户类"
我正在尝试 Flink 的新 Python 流 API，并尝试使用 ./flink-1.6.1/bin/pyflink-stream.sh examples/read_from_kafka.py 运行我的脚本.python 脚本相当简单，我只是尝试从现有主题中使用并将所有内容发送到 stdout(或日志目录中的 *.out 文件，默认情况下输出方法在该文件中发出数据). import glob导
..
Apache Flink:Python 流式 API 中的 Kafka 连接器，“无法加载用户类"
我正在尝试 Flink 的新 Python 流 API，并尝试使用 ./flink-1.6.1/bin/pyflink-stream.sh examples/read_from_kafka.py 运行我的脚本.python 脚本相当简单，我只是尝试从现有主题中使用并将所有内容发送到 stdout(或日志目录中的 *.out 文件，默认情况下输出方法在该文件中发出数据). import glob导
..
PyFlink 向量化 UDF 抛出 NullPointerException
我有一个 ML 模型，它接受两个 numpy.ndarray - users 和 items - 并返回一个 numpy.ndarray predictions.在普通的 Python 代码中，我会这样做: model = load_model()df = load_data() #DataFrame包含4列，分别是user_id、movie_id、 rating、timestamp用户 = d
..
使用 Apache Flink SQL 从 Kafka 消息中获取嵌套字段
我正在尝试使用 Apache Flink 1.11 创建一个源表，我可以在其中访问 JSON 消息中的嵌套属性.我可以从根属性中提取值，但我不确定如何访问嵌套对象. documentation 建议它应该是 MAP 类型，但是当我设置它时，我收到以下错误 : java.lang.UnsupportedOperationException: class org.apache.calcite.
..
Apache Flink:Python 流式 API 中的 Kafka 连接器，“无法加载用户类"
我正在尝试 Flink 的新 Python 流 API，并尝试使用 ./flink-1.6.1/bin/pyflink-stream.sh examples/read_from_kafka.py 运行我的脚本.python 脚本相当简单，我只是尝试从现有主题中使用并将所有内容发送到 stdout(或日志目录中的 *.out 文件，默认情况下输出方法在该文件中发出数据). import glob导
..
在 Java Flink 作业中使用 Python 用户定义函数
无论如何要在Java Flink Job中使用python用户定义的函数，或者无论如何来交流例如flink与java完成的转换结果以及python用户定义的函数来应用一些机器学习的东西: 我知道从 pyFlink 你可以做这样的事情: table_env.register_java_function("hash_code", "my.java.function.HashCode") 但是
..
使用Apache Flink SQL从Kafka消息获取嵌套字段
我正在尝试使用Apache Flink 1.11创建一个源表，在这里我可以访问JSON消息中的嵌套属性.我可以从根属性中删除值，但不确定如何访问嵌套对象. 文档建议它应该是 MAP 类型，但是当我设置它时，出现以下错误 :java.lang.UnsupportedOperationException:类org.apache.calcite.sql.SqlIdentifier:MAP 这
..
在Java Flink作业中使用Python用户定义的函数
无论如何，是否可以在Java Flink作业中使用python用户定义的函数，或者无论如何进行通信，例如通过flink与python用户定义的函数与java进行转换的结果，以应用一些机器学习的东西: 我知道您可以通过pyFlink执行以下操作: table_env.register_java_function("hash_code"，"my.java.function.HashCode"
..
Apache Flink:Python流API中的Kafka连接器，“无法加载用户类"
我正在尝试Flink的新Python流API，并尝试使用./flink-1.6.1/bin/pyflink-stream.sh examples/read_from_kafka.py运行我的脚本. python脚本非常简单，我只是想从一个现有的主题中使用并将所有内容发送到stdout(或默认情况下output方法在其中发出输出数据的日志目录中的* .out文件). import glob i
..