apache-flink 相关问题

Flink Kafka Consumer - 更改超时异常的重试次数

我有一个 Flink 作业，它使用 Flink 的 KafkaSource（“Flink-connector-Kafka”库）使用来自 Kafka 的数据。 有时我会收到超时异常“无法获取所有元数据

如何在 flink-conf.yaml 中的 flink 指标报告器的 filter.includes 参数中包含多个过滤器？

我想了解通用flink指标报告器的工作原理，并且我正在尝试filter.includes参数。如何在 flink-conf.yaml 中为此选项指定多个过滤器？文档

是否可以将 flink 数据流的输出捕获到列表中

我是flink新手，不知道这是正确的方法还是愚蠢的事情，我有一个字符串数据类型的数据流，我正在尝试将数据流中的数据捕获到列表中，我正在尝试一些...

如果链接正在运行，如何在 flink-sql 作业图中可视化运算符执行情况

我有flink-sql应用程序，它只是通过连接多个表来执行简单的简单插入到丰富表中。 创建表T1（...）WITH（'连接器'='upsert-kafka'，'主题'='T1'，...） ...

如何查看我的Task到KeyGroup到flink-sql应用程序中的Key分配是什么

我有flink-sql应用程序，它只是通过连接多个表来执行简单的简单插入到丰富表中。 创建表T1（...）WITH（'连接器'='upsert-kafka'，'主题'='T1'，...） 创建

将数据发送到异步函数后清除状态是否安全？

我想构建一个对数据进行一些聚合的应用程序，并使用计时器将该聚合发送到一个异步步骤，该异步步骤将其转储到其他地方。数据发送后...

Apache Flink GCS FileSink 由于许多小文件而性能不佳

我正在开发一个apache-flink（v 1.17.2）流应用程序，其中我的文件接收器是Google Cloud Storage。性能非常差，我的任务管理器 100% 忙。我相信这是有关系的...

使用并行度 > 1 和 `env.fromElements` 导致应用程序挂起

我使用 JUnit5 运行一个小测试用例 公共类 APipelineTest { @ClassRule 公共静态 MiniClusterWithClientResource flinkCluster = 新的 MiniClusterWithClientResource( ...

在 Apache Flink 作业中设置环境变量

！编辑下面的答案！ 我有 minikube 本地 Kubernetes 集群，并且部署了 Flink Kubernetes Operator。 以下 Flink Kubernetes Operator git 存储库和 Basic HA ex 中提供的示例...

Flink SQL：插入 MongoDB 时不要具体化 NULL 字段

创建表测试 （ _id 字符串主键未强制执行， 一个INT ） 和 （ '连接器' = 'mongodb', 'uri' = 'mongodb://用户:密码@数据库:27017', '数据库'='数据库'，...

在 PyFlink 中访问来自 ADLS 的流数据

尝试从 ADLS 设置 Pyflink 流，当前尝试使用 StreamExecutionEnvironment.from_source() 方法读取 json 文件。 代码如下所示： 来自 flink.plan.Envir...

Flink - 如何触发源刷新

我的程序有两个流 - 一个数据流和另一个丰富流。丰富流被广播并连接到数据流。丰富流偶尔会刷新，但有些...

解决Flink SQL windows tumble性能下降问题

我有一个flink sql应用程序，它将来自Kafka的数据实时分组到数据库中。数据按照时间间隔进行聚合：[1,5,10,30,60min,daily]并写入对应的DB表...

当下游任务并行度不同时，使用 KeyBy 与 reinterpretAsKeyedStream()

我在reinterpretAsKeyedStream的文档中看到这个警告 警告：重新解释的数据流必须已经以与 Flink 的 keyBy 分区方式完全相同的方式进行预分区...

Flink 窗口关闭时间

代码如下。根据我的理解，当水印大于窗口结束时间时，窗口将被关闭。但我的测试结果非常混乱。 公开课测试{ ...

Flink 的 keyBy 操作延迟高

尽管将 setBufferTimeout 设置为 0，但我测量发现大多数情况下 keyBy 操作的延迟通常在 0-1 毫秒之间。然而，随着程序继续运行，我偶尔会观察到

Flink 本地执行

我想知道如何在本地执行模式下运行flink table api。只是一个java程序并使用intelleji运行它，不需要任务管理器和作业管理器等。

Flink：联合多kafka源时为什么会丢失数据？

我是flink的新人。我有五个具有不同数据模式的无限卡夫卡源。我想减少消息并获取最新消息，然后使用相同的密钥外部连接所有 kafka 源。 所以我用un...

flink 1.16 使用 flink-doris-connnect-1.14 包接收 doris，为什么运行缓慢？

我使用的是 Flink 1.16 版本，并通过 Maven 合并了 Flink Doris Connector 1.14 包。当尝试将一批包含 30,000 条记录的数据下沉到 Doris 数据库时，...

我可以在 AWS Managed Apache Flink 上的哪里存储 rockDB 状态？

我正在编写一个 Flink 作业来存储状态并使用 RocksDB 执行聚合，但我不确定在哪里可以让 RocksDB 存储状态。在我的本地计算机上，我将其存储在 ~/Downloads/flink-1.15.4/