问题标签 [pyflink]

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

只是想检查我的代码是否错误或者这是 pyflink 1.11.X

当我尝试通过查询（'GroupBy Aggregation'或'GroupBy Window Aggregation'）计​​算组中元素的数量时，pyFlink会抛出以下错误：

如果我升级到 pyflink 1.12.XI，可以通过将类型转换为 double 来解决。令人惊讶的是，它适用于 1.12，但不适用于 1.11。

有人可以对此有所了解吗？请在下面找到重现错误的示例。

我正在尝试使用 Pyflink 的 JdbcSink 连接到 Oracle 的 ADB 实例。我可以在 Flink 的官方文档中找到使用 java 的 JdbcSink 示例。但是没有为 Python API 提供相同的内容。我试图将 JdbcSink 可以在 java 中实现的方式复制到 python 中，但是 Pyflink 的 JdbcSink 的方法签名与 Java 的 JdbcSink 不同。而且我找不到任何关于使用 Pyflink 的 JdbcSink 的示例或文档。我是 pyflink 的新手。这是我一直在尝试做的事情：

使用 Flink 1.13.1 和一个 pyFlink 和一个用户定义的表聚合函数 (UDTAGG)，并将 Hive 表作为源和接收器，我遇到了一个错误：

这是接收器的 SQL CREATE TABLE

这里有什么问题？

命令：/space/flink/bin/flink run-application -t yarn-application -Dyarn.application.name=kafka-hive --pyFiles /space/testAirflow/airflow/dags/ -py /space/testAirflow/airflow/dags /chloe/kafka_to_hive_1.py

flink 抛出异常：java.lang.IllegalArgumentException：应该只有一个 jar

应用模式不支持pyflink吗？

如文档中所述：

例如，数据管道可能会监视文件系统目录中的新文件并将其数据写入事件日志。另一个应用程序可能将事件流具体化到数据库或增量构建和细化搜索索引。

那么，如何在使用 Flink 的同时关注本地文件系统的文件更新呢？

在这里，该文件还提到：

流式传输的文件系统源仍在开发中。未来，社区将增加对常见流用例的支持，即分区和目录监控。

这是否意味着我可以使用 API 进行一些特殊的流式传输？如果您知道如何使用流式文件系统源，请告诉我。谢谢！

试图将这样的数据（zeek 连接数据）加载到 pyflink。我的问题是名称带有点的 id 字段，因为它们最初是 zeek 中的元组。

我将不胜感激有关如何执行此操作的任何帮助。

我有这个玩具管道

运行得很好，但是每次我尝试取消对映射阶段的注释时，无论是虚拟内联 lambda 还是 MapFunction，它都会爆炸说：

我正在使用 pyenv 3.8 解释器，有人知道基本数据源和输出是如何运行的，但地图显示 pyflink 模块丢失了吗？

附录：这只发生在 PyCharm 上，当我在控制台上运行脚本时不会发生，所以我担心它是 PyCharm 和 pyenv 之间的东西

我有一个看起来像这样的模式：表：org_table

当我像这样查询此表时：

在表上运行上述查询时，出现错误。我在这里缺少什么吗？

最新稳定版V1.13.2的Python API是否支持DataStream和Table之间的转换？