我们有一个flink流应用程序，从卡夫卡读取消息。由于某些原因，我们不得不从kafka重置命令将kafka偏移重置为最新，因为有大量堆积。我们希望flink应用程序跳过所有这些消息，并从重置后出现的新消息开始。问题是因为flink在内部管理它的偏移量，它不知道这个重置，它现在只从后向读取msg (重置前的偏移点)，现在也不能提交偏移量。因此，每次重启flink应用程序时，它都会再次从同一点读取。因此，我们在每次重启时
浏览 62提问于2021-06-19得票数 4
回答已采纳
2回答
我正在构建一个卡夫卡-> Flink ->卡夫卡管道，它可以处理划定的“会话”数据。实现非常简单--由session_key实现，将start_event存储到ValueState中，将条目存储到ListState中，然后当end_event到达时，对所有事件运行处理逻辑，并将结果推入输出Kafka主题。我应该如何正确地从这件事中恢复过来？如果卡夫卡偏移量已经被提交，并且没有任何end_event将它从卡夫卡中为该session_key，我
浏览 7提问于2022-01-30得票数 0
回答已采纳
在应用程序非常新的开始时，必须从检查点上卡夫卡主题的最新偏移量中读取，它必须在重启后(当应用程序手动/系统错误终止时)将所消耗的偏移量提交给Kafka ，它必须从上次提交的偏移量中选择，并且必须使用消费者延迟使用新的KafkaConsumer API (KafkaSource)，我面临以下问题
能够执行上述要求，但不能在检查点(500 on )上提交所消耗的偏移量。而是在
浏览 10提问于2021-12-16得票数 2
1回答
假设我的主题中已经有10个数据，现在我开始编写Flink，消费者将使用第11个数据。
}和pom.xml： <groupId>org.apache.<em
浏览 1提问于2018-10-31得票数 2
回答已采纳
2回答
如果我在答案中遗漏了什么，那么请重新表述答案，因为我显然遗漏了一些东西。 "org.apache.flink" %% "flink-connector-kafka-0.9" % "1.1.2"
浏览 4提问于2016-09-12得票数 7
回答已采纳
我在读书启用检查点:如果启用了检查点，则Flink消费者将在检查点完成时提交存储在检查点状态中的偏移量。这确保了Kafka代理中提交的偏移量与检查点状态中的偏移量一致。用户可以通过调用使用者上的setCommitOffsetsOnCheckpoints(布尔值)方法来选择禁用或启用偏移提交(默认情况下，行为为真)。请注意，在此场景中，属性中的自动周期
浏览 0提问于2021-06-25得票数 0
回答已采纳
1回答
我有一个Flink流应用程序，它正在消耗来自Kafka主题的数据，这个主题有3个分区。尽管应用程序一直在运行和工作，并且没有任何明显的错误，但我看到flink应用程序在所有3个分区上的消费者组都出现了滞后。Offset:9561Offset:9563Offset:9565
日志端偏移值为9568，当前偏移量为9566.为什么这些补偿在控制台使用者中不可用，为什么存在这种滞后？例子-
浏览 7提问于2022-06-17得票数 1
1回答
以下是我的假设：
Flink从kafka消耗一些固定数量的事件(一次从多个分区获得多个偏移量)，并等待，直到它到达接收器，然后是检查点。在成功的情况下，它提交kafka分区，偏移它读取，并维护一些与它正在写入的hdfs文件相关的状态。在kafka将事件交给其他运营商之后，进行了多个转换，而kafka消费者则处于闲置状态，直到它得到它发送的事件的成功确认。在失败的情况下，flink转到以前的检查点状态
浏览 0提问于2020-03-12得票数 0
我试图在Exactly-Once集成中实现Flink-Kafka语义。我将按照第一个用户提到的在使用者端添加一个属性。我是否也应该在生产者端添加Idempotent属性？我的Flink版本是1.13.5，Scala 2.11.12和我正在使用Flink Kafka connector 2.11。我认为我无法使用EXACTLY_ONCE提交事务，因为检查点不是在上述路径上编写的。附加Web的屏幕截图：
我需要为此设置
浏览 1提问于2022-01-07得票数 1
回答已采纳
在从Kafka摄取消息的Flink流应用程序中，1)如何禁用自动提交? 2)在成功处理消息后如何从Flink手动提交？ 谢谢。
浏览 138提问于2019-02-08得票数 1
我遇到的问题是，在重新启动Kafka和Flink作业时，FlinkKafkaConsumer偏移被重置为0，因此数据被重新处理，即使我启用了检查点，并且在Flink作业中启用了一次语义。这是我的环境细节
在KubernetesKafka源主题下运行的有10个分区，没有replication.Kafka，有3个代理。启用Flink检查点时，semantics.Flink版本为1.12Flink通过负载
浏览 6提问于2022-01-25得票数 0
使用来自Kafka主题的数据，Flink和SparkStreaming都提供了检查点机制，前提是auto.commit.enabled设置为false。火花博士说：
在启用了Flink的检查点之后，Flink Kafka消费者将使用来自某个主题
浏览 6提问于2017-12-12得票数 1
1回答
简而言之，我想从一开始就对Kafka 中的数据重新运行Flink管道。我尝试过auto.offset.reset = smallest/earliest和更
浏览 0提问于2016-02-20得票数 3
2回答
Flink不对kafka进行抵消
、、、、
我有一个flink流媒体工作，是从卡夫卡读取数据，并只是记录它。我启用了检查点。任何帮助都是徒劳无功的。{$KAFKA_HOME/bin/kafka-consumer-groups.sh --new-consumer --bootstrap-server localhost:9092 --describe--group flink-consumer-group
Error while executing consumer
浏览 5提问于2017-02-06得票数 3
回答已采纳
我创建了一个简单的Apache光束流管道，它从Kafka读取数据，进行一些处理，并通过调用一些外部服务的API来持久化结果。我希望确保在管道重启或失败期间不会丢失数据，因此我希望在特定doFun执行结束时成功调用接口后，手动将记录偏移量提交给Kafka。在我之前的Kafka经验中，我知道通过使用Kafka Consumer的below接口，我可以手动将记录偏移量<
浏览 10提问于2021-09-21得票数 1
回答已采纳
1回答
下面的Flink流应用程序在本地运行，使用SQL编写： val targetPath = f"file://${new java.io.File(".").getCanonicalPath}/kafka-to-fs-csvBIGINT METAD
浏览 5提问于2022-03-14得票数 0
回答已采纳
我们有一个Flink作业，它具有以下拓扑结构：我们在接收器操作符open-override函数中设置了一个活动(就绪)状态。
浏览 4提问于2022-09-19得票数 0
回答已采纳
我有一个Flink作业来消费一个Kafka主题，并将其沉没到另一个主题，Flink作业设置为auto.commit，间隔为3分钟(检查点禁用)，但在监控端，有3分钟的延迟。但我们希望在没有3分钟延迟的情况下实时监控处理，因此我们希望有一个功能，即FlinkKafkaConsumer能够在接收函数后立即提交偏移量。
有没有办法在Flink框架中实现这个目标？在第53行，我试图创建一个KafkaCons
浏览 0提问于2020-12-29得票数 0
Flink版本: v1.15.2I有一个应用程序，它是卡夫卡的使用者，它使用使用者组'TopicA‘使用主题'topic_a'中的数据。有一个Flink任务，其卡夫卡源表使用的卡夫卡消费者组也是“TopicA”，但使用的是主题“topic_b”的数据
此时，
浏览 10提问于2022-09-23得票数 0
回答已采纳
云服务器
ICP备案
云直播
对象存储
腾讯会议
腾讯云开发者
