Apache Flink是一个用于可扩展批处理和流数据处理的开源平台。 Flink在一个系统中支持批量和流分析。分析程序可以用Java和Scala中简洁优雅的API编写。
在 Flink 的 Elasticsearch Sink (Python) 中为嵌套 JSON 定义正确的数据类型
我正在使用 Apache Flink 并尝试使用内置的 Elasticsearch Sink 将文档接收到 Elasticsearch 中。我很难定义接收器接受的正确数据类型...
CustomDeserializer 无法在 pyflink 中接受字符串作为元组
我收到了 Kafka 数据，看起来像（“field1”，“field2”）。所以，现在我尝试使用自定义反序列化器通过 Pyflink 1.17.1 解析此输入，它看起来像贝尔...
使用 ND4J 和 Flink，我有一个处理函数，它接收 POJO，使用 linalg ndarray 使用一堆数学计算结果，并输出一个 pojo。 在集群上运行程序时，
我有一个将数据接收到 Kafka 主题的流作业，还有另一个使用 kafka 主题中的数据的流作业。我的Kafka集群的分区数配置为3。 当我设置
我有一个 json 部分，其中包含以下内容： {"tokenType":"电子邮件","tokenList":["token1","token2","token3","token4"]} 我有一个航班...
在PyFlink中使用SourceFunction和SinkFunction
我是 PyFlink 的新手。我已经用Java完成了官方培训练习：https://github.com/apache/flink-training 然而，我正在进行的项目必须使用Python作为编程语言。我...
我在将 flink 应用程序部署到 AWS 帐户时遇到以下错误。 java.lang.NoSuchMethodError：'org.apache.flink.metrics.MetricGroup org.apache.flink.api.common.functions.RuntimeContext。
我有一个简单的类，如下所示： 类我的类{ 私有静态 SomeClass 对象 = null; 公共无效初始化（SomeclassinjectedObject）{ 对象=注入对象； } } 我在我的...
使用 Flink (Pyflink) 聚合后数据未发送到 Kafka 主题
我正在研究一个简单的数据聚合示例。它只是按记录 ID 进行分组并计算 1 秒滚动窗口内的平均值。 使用 flink faker 连接器生成源数据...
为什么我的 Flink 任务管理器在将 MySQL 数据传输到 Iceberg 时崩溃？
源表大约有 4TB 大。所以我们尝试使用Flink来流式迁移它们。 理想情况下，Flink 将从 MySQL 中选取数据并将其逐行传输到 Iceberg。每行插入后我...
MetricQueryService - 某些指标将不会被报告
升级Flink v.1.16.1（从v.1.13.2）后，我看到以下日志： 2023-03-06 INFO org.apache.flink.runtime.metrics.dump.MetricQueryService [] - 某些指标将不会被报告...
Jobmanager重启后Flink无法从checkpoint恢复
我正在 Docker-compose 上运行 flink 集群，其中包含 1 个 jobmanager 和 1 个 taskmanager。我通过重启Jobmanager的容器来测试checkpoint的机制。但我发现状态没有恢复
我们有多个Flink流应用程序。这些应用程序每 30 秒左右向 Kafka 提交一次偏移量。我们看到 Flink 根本没有向 Kafka 提交偏移量。这会导致不匹配...
我们如何配置 Flink 应用程序以仅启动/重新启动崩溃的 Pod/（子）任务，而不是重新启动整个作业，即重新启动作业/管道中的所有任务/子任务，包括...
我有一个 Kafka 生产者，它从两个大文件中读取数据并以具有相同结构的 JSON 格式发送它们： def create_sample_json(row_id, data_file): return {'row_id':int(row_id), '行...
Apache Flink dataStream.sinkTo() 不接受 KafkaSink<String> 作为参数。正在等待Sink<String, ?, ?, ?>
我是 Apache Flink 的新手。我正在尝试从 Kafka 流式传输数据，在 Flink 上执行某些操作并将数据发布到 Kafka 中的其他某个主题。 下面是添加的依赖项 ...
使用flink sql join 2 source时如何读取rocksdb状态
我的sql定义为 如果 TABLE_1 不存在则创建表（ 标头 VARCHAR NOT NULL， id VARCHAR 不为空， `时间戳` TIMESTAMP_LTZ(3) NULL, 类型 VARCHAR NOT ...
新的状态序列化器不得与 Apche Flink SQL 中的旧状态序列化器不兼容
我正在使用 Kafka 从 Flink Datastream 读取数据（从多个流读取），然后我使用 flink SQL 将这些流转换为临时视图并执行联接操作。欲加盟
如何在 PyFlink 中为 CassandraSink 提供凭证
请告诉我，也许有人遇到过这种情况。 我需要在 PyFlink 中配置 CassandraSink 以便将聚合数据写入流上的 Cassandra/Skylla 表。我找不到...
有人在 Mac m1 机器上构建 PyFlink Docker 容器时遇到同样的问题吗？ 在 apache-flink 和 apache-flink-