apache-flink 相关问题

Flink的map功能好像是以批处理模式运行的

我是Flink（使用python）的新手，最近我遇到了一个问题，总之我相信（实际上我已经验证了这一点）即使我在流模式中设置了环境，map函数也会以批处理模式运行.. .

分区特定的flink kafka消费者

我创建了一个流程序，用于使用flink和kafka流式传输mongodb oplog。根据与 Flink 支持团队的讨论，无法通过 kafka 分区保证流的排序...

如何使用kafka主题在FlinkSQL中运行批处理模式

我有创建语句FLINK SQL的记录。 创建表 en_trans ( `transid` 整数， `产品名称` 整数， 主键 (transid) 未强制执行 ） 和 （ '连接器' = 'upsert-kafka', '

有没有办法覆盖Flink /lib 文件夹中的一些已发布的函数？

我的 Flink 应用程序是由 Yarn 部署的，我将所有必需的 Flink 依赖项包含在 Flink 的 /lib 文件夹中。现在我想向 jar 'flink-table-planner' 添加一些独特的功能，但我...

为什么 Flink 在每次调用测试工具中的 processElement 后都会重置我的状态？

我正在使用 Flink 的 KeyedOneInputStreamOperatorTestHarness 并调用 processElement 两次。 processElement 将更新状态以计算所看到的元素数量。 在这种情况下，拨打电话后

将相同的方法引用传递给 apache flink 中的过滤器会抛出 classcastException

以下代码抛出 ClasscastException (java.lang.ClassCastException: class java.lang.Integer无法转换为 class java.lang.String ) 最终数据流源 st1 =

无法运行基本的 PyFlink 示例

我有这个玩具管道 从 pyflink.datastream 导入 StreamExecutionEnvironment def 管道(): # 创建环境 env = StreamExecutionEnvironment.get_execution_environment() 环境

Zeppelin 笔记本中出现错误，指出无法创建接收器

我正在尝试在 Apache Flink 中处理后从 1 个 KDS 写入另一个 KDS。我使用 Zeppelin 笔记本通过以下查询创建接收器表： %flink.ssql 创建表 seller_revenu...

Flink 与 Kafka Source 和 Iceberg Sink 不写

使用Flink我尝试从Kafka读取数据，将Protobuf事件转换为Json字符串并将其写入Iceberg中的表中。 我按照官方文档编写了代码，但我必须有

Flink 可以使用事务绑定写入多个 Kafka 主题吗？

有没有一种方法可以使用 KafkaSink 将记录接收到多个主题以确保交易？该文档没有提及类似的内容。 有一个 stackoverflow 线程建议

Flink Stream 处理处理部分失败并避免重新处理

我有 Flink 流处理应用程序，它从 Pulsar Topic 读取消息流，处理它们并将文件存储在 S3 中。它执行以下操作。 每隔一段时间阅读 Pulsar 主题的消息...

在 KeyedProcessFunction 类的构造函数中设置类成员值不起作用

我有一个扩展 Apache Flink 中的 KeyedProcessFunction 的类。 该类有一个临时成员，例如 SomeType ok 在该类的构造函数中，类成员 ok 是 populatio...

Flink - 公开新的自定义 REST API

在 Apache Flink REST API 文档的“开发”部分中，写道： 要添加新请求，需要 添加一个新的 MessageHeaders 类作为新请求的接口， 添加...

当sink发生异常时，source算子在外部存储中的状态是回滚吗？

在下面的Flink逻辑中，我想知道检查点的机制。 假设有如下简单的数据流。 源（kafka）->进程->接收器（db） 据我所知，

Flink postgres jdbc 源连接器读取大写字段失败

我尝试使用flink1.18 sql-client.sh读取pg数据，但失败，字段为大写，名为idLog。 我发现flink会自动将大写字段转换为小写字段。 我努力工作

Apache Flink 增加并行度并不能提高性能

我们有一个规则流，必须对存储在文件中的 n 条记录运行。下面给出的是我编写的 Flink 作业的伪代码。 溪流 .flatMap( // 1. 阅读...

现在批处理的意义是什么？

我有批处理背景，但最近开始专门从事流处理工作。我发现在 apache Spark 结构化流中，例如，我们可以执行 Trigger.AvailableNo...

了解水印

我只是想表达我对 BoundedOutOfOrder Watermarks 在 FLINK 以及任何流处理框架中如何工作的理解。 事件处理顺序： 11:00 11:01 11:0...

Flink - Kryo 序列化器 scala 扩展不可用

我在我的应用程序中使用flink，并且在事件反序列化期间出现以下错误 {“level”：“INFO”，“logger”：“org.apache.flink.api.java.typeutils.runtime。

在 Flink 的 Elasticsearch Sink (Python) 中为嵌套 JSON 定义正确的数据类型

我正在使用 Apache Flink 并尝试使用内置的 Elasticsearch Sink 将文档接收到 Elasticsearch 中。我很难定义接收器接受的正确数据类型...