PyFlink SQL本地测试

文章/答案/技术大牛

发布

1回答

apache-flink、flink-streaming、flink-sql、pyflink

所以我有一个用PyFlink SQL API编写的简单的聚合作业。该作业从AWS Kinesis读取数据并将结果输出到kinesis。我很好奇是否可以用pytest对我的流水线进行单元测试？但是我如何创建一个本地Flink会话来在pytest中运行作业呢？我们这里有没有最佳实践推荐？谢谢!

浏览 35提问于2021-11-12得票数 0

回答已采纳

1回答

在一个独立的回购系统中是否有PyFlink SQL单元测试的例子？

python、apache-flink、flink-sql、pyflink

是否有一个自包含存储库的例子显示如何执行PyFlink的SQL单元测试(如果可能的话，具体而言是1.13.x )？有一个相关的问题是，建议使用来自PyFlink本身的一些测试。我遇到的问题是，PyFlink回购假设在Java上有很多东西，并且有一些PyPi实用类是可用的(它们不是通过PyPi apache分发的)。复制一个示例单元测试(，如相关的SO问题所建议的)。当我尝试运行测试时，我会得到一个错误，因为测试用例无法确定要下载哪个

浏览 4提问于2022-04-23得票数 0

1回答

Zeppelin Python Flink无法打印到控制台

apache-flink、apache-zeppelin、amazon-kinesis-analytics

非常简单的代码：from pyflink.common.typeinfoimport Typesfrom pyflink.datastream.connectorsor StreamExecutionEnvironme

浏览 20提问于2022-03-13得票数 1

1回答

单元测试PyFlink UDF失败

apache-flink、pyflink

我正在使用PyFlink，我想对用Python编写的UDF进行单元测试。要测试以下简单的udf：from pyflink.table import DataTypesdef add(i, j):我创建了一个

浏览 4提问于2021-03-23得票数 2

回答已采纳

2回答

pyflink kafka连接器将接收到的json数据反序列化为null

json、apache-kafka、apache-flink、stream-processing、pyflink

我正在使用PyFlink创建一个流处理器。当我将Kafka连接到Flink时，一切工作正常。但是当我向kafka发送json数据时，PyFlink接收到它，但反序列化程序将其转换为空。PyFlink代码是 from pyflink.common.serialization import Encoderfrom pyflink.common.serialization i

浏览 286提问于2021-06-13得票数 0

1回答

是否有方法可以将Azure SQL server用作数据源？

apache-flink、pyflink

我的项目目前构建在Azure上(数据存储在Azure SQL server中)，我目前正在尝试通过利用PyFilnk将流/批处理功能引入到我的项目中。但是，我没有找到任何关于如何将PyFlink连接到Azure SQL server的文档，有没有一种方法可以在PyFlink中使用Azure SQL server作为数据源？

浏览 13提问于2021-07-22得票数 0

1回答

在PyFlink中导入本地包

python、apache-flink、pyflink

我正试图在PyFlink项目中编写一个本地包。但我只能通过相对路径导入。喜欢是否可以在导入为PyFlink的env.add_python_file('/path_to_project')项目中的包中使用绝对路径？

浏览 6提问于2021-11-18得票数 0

回答已采纳

1回答

Apache 1.11无法在SQL函数DDL中使用Python

python、apache-flink、user-defined-functions、pyflink、flink-table-api

根据这一汇合点页面：python位于Flink 1.11中，可用于SQL函数中。ci.apache.org/projects/flink/flink-docs-stable/dev/table/sqlClient.html 并在终端上尝试此

浏览 1提问于2020-07-09得票数 0

1回答

PyFlink 1.11.2在注册自定义UDF函数时无法配置[taskManager.Memy.task.off-heap.size]属性

apache-flink、pyflink

我在pyflink非常新，并且尝试使用python注册一个自定义的UDF函数。目前，我在服务器env和本地IDE环境中都遇到了一个问题。当然，我已经将必需的属性添加到我的flink-conf.yaml中，并检查pyflink-shell.sh是否使用指定的配置初始化env，但是这没有任何意义，而且我仍然有一个错误。下面是我正在尝试运行的一个代码示例：from pyflink.table import Bat

浏览 5提问于2020-10-12得票数 0

1回答

pyflink配置错误SQL解析失败

apache-flink、flink-streaming、pyflink

为了在我的应用程序中使用docker下载pyflink版本的1.10，我使用了以下代码片段： image: pyflink/playgrounds:1.10.0import StreamExecutionEnvironmentfrom pyflink.tableExecutionEnvironment from pyflink.table i

浏览 21提问于2021-07-20得票数 0

3回答

Flink Python Datastream API Kafka使用者

python、apache-kafka、apache-flink、stream-processing、data-stream

我添加了在上可用的flink-sql连接器。有人能帮我举个合适的例子来做这个吗？以下是我的代码import os from pyflink.datastreamimport StreamExecutionEnvironmentfrom pyflink.common.

浏览 23提问于2022-02-03得票数 0

1回答

当运行下面的pyflink代码时，我会得到这个错误

python、apache-flink、flink-streaming、flink-sql、pyflink

这是使用apache (Pyflink)从kafka源代码中计算每个chx平均值的代码，我想我已经导入了所有必需的库。在运行代码时，我会得到这个错误from pyflink.table import TableEnvironment, EnvironmentSettingsfrom pyflink.datastream import StreamExecutionEnvironment, TimeCharacteristic from pyflink.table imp

浏览 17提问于2022-04-29得票数 0

回答已采纳

1回答

得到"pyflink.util.exceptions.TableException: findAndCreateTableSource失败“。运行PyFlink示例时

apache-flink、pyflink

我运行在PyFlink程序下面(从复制)from pyflink.table import TableConfig程序，使之更喜欢SQL而不是Table，但我发现它不起作用。from pyflink.table.descriptors import Schema, OldCsv, FileSystem from pyflink.table.expressions importconnect

浏览 2提问于2021-03-15得票数 2

回答已采纳

1回答

Pyflink查看木星笔记本中的数据，比如头/在熊猫中显示

jupyter-notebook、apache-flink、flink-sql、pyflink

我是PyFlink的新手，正在努力学习。我使用pyflink和jupyter笔记本并执行一些基本操作。因此，当我执行某些操作时，它会返回内存中的一个表位置，如"<pyflink.table.table.Table at 0x1eeffd773a0>。要查看数据，我需要调用to_pandas()方法。Pyflink中是否有任何方法可以查看处理过的数据，比如熊猫的head，或者在火火花中显示/显示。“

浏览 5提问于2022-01-30得票数 0

回答已采纳

1回答

如何将一些外部jars库配置到flink docker容器

docker、containers、apache-flink

:/usr/local/lib/python3.7/site-packages/pyflink/lib expose: - "6123lib/flink-dist_2.12-1.11.0.jar::: 从日志中，您可以看到一些默认库已加载到系统中，但我希望添加一些jars，如flink-jdbc_2.11-1.9.0.jar，它位于我的本地文件系统中我的本地j

浏览 121提问于2020-10-11得票数 1

1回答

如何向远程Kubernetes会话集群提交pyFlink作业？

python、kubernetes、apache-flink、pyflink

目前，我有一个正在运行的Flink Kubernetes会话集群(Flink版本1.13.2)，并且我可以通过port-forward访问web，我也可以通过这个命令从本地环境提交WordCount jar但是，当我试图通过命令pyFlink提交.我尝试过许多方法，包括创建virtualenv、传递pyClientExecutable和pyexec、同步本地和远程python版本，但它们都没有工作。我遗漏了什么？注意事项：当我在作业管理器pod中提交pyFlink word_count示例时，它运行时没有任何问题。

浏览 4提问于2021-10-16得票数 2

1回答

使用Apache-flink将处理过的流数据存入数据库

python、apache-flink、flink-streaming、pyflink

是否可以使用pyflink将处理过的流数据存入数据库？所有写入已处理数据的方法都被限制为以txt、csv或Json格式保存它们，并且没有办法将数据与数据库一起接收。

浏览 1提问于2020-07-27得票数 2

2回答

flink run -py与python run的区别

python、apache-flink、pyflink

我最近在学习pyflink，但我有点困惑。我们知道pyflink table API将流/批处理转换为表，并在其上做一些工作，最后到达您想要的位置。evnironment_settings = env_settings) 但是我发现其中一个样式只适用于flink run -py batch.py，其他一些样式可以作为python batch.py运行在本地executionEnvironment和BatchTableEnvironment/StreamTableEnvironment有什么区别，什么时候可以使用flink

浏览 107提问于2020-12-18得票数 0

1回答