在Kafka中添加重新分区后丢失旧的聚合记录

在Kafka中添加重新分区后，可能会导致旧的聚合记录丢失。这是因为重新分区会改变消息的分布情况，原本存储在某个分区的消息可能会被重新分配到其他分区，从而导致无法再访问到这些消息。

为了避免丢失旧的聚合记录，可以考虑以下几个解决方案：

备份数据：在进行重新分区之前，先对旧的聚合记录进行备份。可以将这些记录导出到其他存储系统（如数据库）中，以便后续需要时进行恢复。
使用Kafka Streams的状态存储：Kafka Streams是Kafka提供的一个流处理框架，它可以帮助我们进行聚合操作。在使用Kafka Streams时，可以将聚合结果存储在状态存储中，而不是直接存储在Kafka的分区中。这样，在重新分区后，聚合结果仍然可以从状态存储中恢复。
使用Kafka Connect进行数据同步：Kafka Connect是Kafka提供的一个工具，用于将Kafka与其他数据存储系统进行连接。可以使用Kafka Connect将聚合记录同步到其他存储系统中，以便在重新分区后进行恢复。
使用Kafka的日志压缩功能：Kafka提供了日志压缩功能，可以将消息进行压缩存储。在重新分区之前，可以先对旧的聚合记录进行压缩，以减少存储空间的占用。这样即使在重新分区后，这些压缩的记录仍然可以被保留。

总之，在进行重新分区操作时，需要考虑数据的备份和恢复策略，以确保旧的聚合记录不会丢失。具体的解决方案可以根据实际情况和需求进行选择和调整。

腾讯云相关产品：腾讯云消息队列 CKafka 产品介绍链接地址：https://cloud.tencent.com/product/ckafka

Kafka流状态存储区rocksdb文件大小在手动删除消息时不会减少

apache-kafka-streams、rocksdb、spring-cloud-stream-binder-kafka

我使用处理器api从状态存储中删除消息。Delete是成功的，我通过使用由kafka键对状态存储的交互式查询来确认，但是它并没有减少本地磁盘上tmp/ kafka -streams目录下的kafka流文件大小。 @Override public void init(ProcessorContext processorContext) { this.processorContext = processorContext; processorContext.schedule(Duration.ofSeconds(10), PunctuationType.ST

浏览 7提问于2020-02-12得票数 0

回答已采纳

2回答

卡夫卡的内部主题是什么？

apache-kafka、apache-kafka-streams

我们使用kafka流api进行聚合，在其中我们也使用group。我们还使用状态存储来保存输入主题数据。我注意到卡夫卡内部创作了三种主题 Changelog-<storeid>-<partition> Repartition-<storeid>-<partition> <topicname>-<partition> 我无法理解的是当我拥有<topic>-<partition>中的所有数据时，它为什么会创建changelog主题重新分区主题是否包含分组后的数据。

浏览 0提问于2019-05-10得票数 4

1回答

为什么我要用Kafka流来配置一个州立商店？

apache-kafka、apache-kafka-streams、rocksdb

目前，我有以下设置： StoreBuilder storeBuilder = Stores.keyValueStoreBuilder( Stores.persistentKeyValueStore("kafka.topics.table"), new SomeKeySerde(), new SomeValueSerde()); streamsBuilder.addStateStore(storeBuilder); final KStream<byte[], SomeClass> requestsStream = streamsBuilde

浏览 4提问于2019-12-07得票数 2

回答已采纳

2回答

为什么源主题分区发生变化时，kafka streams线程会死掉？有谁能指出这方面的阅读材料吗？

java、apache-kafka、apache-kafka-streams

我们增加了用于并行处理消息的分区数量，因为消息的吞吐量很高。一旦我们增加了分区的数量，订阅该主题的所有Streames线程就会死掉。我们更改了消费者组id，然后重新启动了应用程序，它工作正常。我知道应用程序的changelog topic的分区数应该与源topic相同。我想知道这背后的原因。我看到了这个链接- https://issues.apache.org/jira/browse/KAFKA-6063?jql=project%20%3D%20KAFKA%20AND%20component%20%3D%20streams%20AND%20text%20~%20%22partition%

浏览 23提问于2019-02-12得票数 8

回答已采纳

1回答

使用自定义密钥创建KTable时出错

java、apache-kafka、apache-kafka-streams

Use-case -有一个包含消息的主题(null，元数据)。我需要从主题创建一个Ktable，其中键(metadata.entity_id)和值作为元数据。该表稍后将用于与具有相同键的流进行连接。 private final static String KAFKA_BROKERS = "localhost:9092"; private final static String APPLICATION_ID = "TestMetadataTable"; private final static String AUTO_OFFSET_RES

浏览 34提问于2020-04-29得票数 1

回答已采纳

2回答

应用程序重新启动后，GlobalKTable是否会保留数据？

apache-kafka、apache-kafka-streams、spring-kafka、spring-cloud-stream

我正在使用Spring Cloud Streams，我有一个接收KStream和GlobalKTable的BiFunction。我不想在应用程序重新启动后丢失GlobalKTable数据，但这正是正在发生的事情。 @Bean public BiFunction<KStream<String, MyClass1>, GlobalKTable<String, MyClass2>, KStream<String, MyClass3>> process() { ... } 我还配置了"materializedAs“属性： spring.clou

浏览 3提问于2021-05-14得票数 0

1回答

是否可以在不丢失数据的情况下更改Kafka主题配置？

apache-kafka

我正在尝试更改Kafka主题配置，我可以通过更改配置来保留我的数据和偏移数据吗？我想知道当我将分区编号1改为3(或将压缩类型gzip改为lz4)或在删除后重新创建时，消息和偏移量信息是否被删除？

浏览 12提问于2021-01-29得票数 0

1回答

在Kafka之上构建过程时，在不影响恢复的情况下最小化故障

apache-kafka、system-design、saga、disaster-recovery

我正在使用一个使用卡夫卡信息的微型服务。它对消息进行一些处理，然后将结果插入数据库中。直到那时，我才向卡夫卡承认了这一信息。要求我将数据丢失保持在最低限度，但恢复速度很快(避免重新处理消息，因为这很昂贵)。我意识到，如果出现某种故障，比如我的微服务崩溃，我的消息就会被重新处理。因此，我想在我的进程中添加某种“检查点”，方法是将转换后的消息状态写入文件，并在失败后读取。我认为这意味着我可以把我的卡夫卡承诺转移到一个更早的阶段，只有在写到文件之后才是成功的。但是，经过进一步的思考，我意识到如果文件系统出现故障，我可能找不到我的文件，例如使用云文件服务可能仍然有失败的机会，即使市场占有率是&g

浏览 5提问于2022-07-30得票数 2

1回答

卡夫卡与动物园管理员的关系

java、apache-kafka、apache-zookeeper

据我所知，动物园管理员存储有关kafka的元数据，主要涉及领导选举、分区细节等。但是当我删除/tmp/kafka-logs目录并重新启动kafka时，它再次在/tmp/kafka-logs目录中获取数据。是否有可能动物园管理员也持有数据，并将其传递给卡夫卡来同步？我使用的是卡夫卡的最新版本，是1.0.1

浏览 1提问于2018-03-28得票数 1

回答已采纳

1回答

Storm的容错性:当工人死亡时数据丢失了吗？

apache-storm、fault-tolerance

我有一个关于容错的问题。考虑到您给出的单词count= ing拓扑，螺栓"WordCount“可能有多个任务，而"fieldsGrouping”用于确保始终将相同的单词分配给相同的任务。我的问题是，如果一些任务死了怎么办？据我所知，storm会尝试重启这些任务。但是当重新启动时，存储在这些任务中的字数也会丢失。这是否意味着在其他或恢复的任务中，这些单词将从0开始？

浏览 1提问于2016-02-19得票数 0

2回答

用于Azure Blob存储的Kafka连接器

apache-kafka、azure-blob-storage、apache-kafka-connect

我需要将推送到Kafka的消息存储在一个很深的存储中。我们正在使用Azure云服务，所以我认为Azure Blob存储可能是更好的选择。我想使用Kafka Connect的接收器连接器API将数据推送到Azure Blob。Kafka文档大多建议HDFS导出数据，但是，在这种情况下，我需要一个运行Hadoop的Linux VM，我想这将是昂贵的。我的问题是Azure Blob存储是存储JSON对象的合适选择，构建自定义接收器连接器对于这种情况是合理的解决方案吗？

浏览 19提问于2016-08-10得票数 3

回答已采纳

1回答

InvalidStateStoreException:州立商店在卡夫卡流中不开放

apache-kafka、apache-kafka-streams

StreamsBuilder builder = new StreamsBuilder(); Map<String, ?> serdeConfig = Collections.singletonMap(SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl); Serde keySerde= getSerde(keyClass); keySerde.configure(serdeConfig,true); Serde valueSerde = getSerde(valueClass); valueSe

浏览 0提问于2019-07-25得票数 2

回答已采纳

1回答

将Ktable按键分组后无效(负)时间戳

apache-kafka、apache-kafka-streams

我正在使用KakfkaStreams (2.3.0)来使用流和Ktable (压缩主题)之间的连接来丰富价值。问题是压缩后的主题被写入到使用流主题的不同分区器中，因此联接无法按预期工作(一些键不匹配，因为它们位于不同的分区中)。代理版本为0.10.2。我开始研究如何重新划分压缩的主题，方法是使用一个groupBy()，后面跟着一个()，但是当它开始读取创建的重新分区主题时，它开始抛出一个带有消息的：输入记录ConsumerRecord(主题=映射-表-重新分区，分区= 18，leaderEpoch =空，偏移量= 0，CreateTime = -1，序列化键大小= 37，序列化值大小=

浏览 1提问于2019-08-28得票数 0

回答已采纳

1回答

Kafka流:窗口时对旧数据的再处理

apache-kafka、apache-kafka-streams

有一个Kafka流应用程序，它通过流连接执行窗口(使用原始事件时间，而不是挂钟时间)，例如1天。如果打开这个拓扑，并从一开始就重新处理数据(就像在lambda风格的体系结构中那样)，那么这个窗口会将旧数据保存在那里吗？例如:如果今天是2022-01-09，而我正在接收2021-03-01的数据，这个旧数据会进入表吗，还是会从一开始就被拒绝？在这种情况下-可以采取什么策略来重新处理这些数据？使用Kafka Streams 2.5.0进行更新

浏览 6提问于2022-01-09得票数 0

1回答

Kafka流拓扑优化

apache-kafka-streams

在为拓扑光学化做准备时，我偶然发现了以下内容：目前，Kafka流在启用时执行两种优化： 1-源KTable重新使用源主题作为变更主题。 2-在可能的情况下，Kafka流将多个重新分区主题折叠为一个单独的重新分区主题。这是第一点的问题。我不完全明白在这里发生了什么。只是为了确保我没有在这里做任何假设。有人能解释一下，以前的状态是什么？ 1- KTable是否使用内部变更主题？如果是的话，谁能带我去找医生吗？接下来，这个变化主题是什么？这是真正的重新插入日志，合并更新操作吗？ 2-如果我最后的猜测是正确的，我不明白为什么由重新插入组成的变化量只能被源主题所取代？

浏览 0提问于2019-07-23得票数 4

回答已采纳

1回答

Kafka Streams状态存储备份主题分区策略

apache-kafka、kafka-producer-api、apache-kafka-streams

Kafka保证具有相同key的消息将始终转到相同的分区。例如，我有一个字符串key为2329的消息。和两个主题t1和t2。不出所料，当我执行此消息的write操作时，它会同时进入两个主题中的分区1。现在问题本身:我使用的是Kafka Streams 0.10.2.0持久状态存储，它会自动创建一个备份主题。现在，在使用键2329的备份主题消息进入另一个分区(分区0)的情况下，这对我来说很奇怪。有没有人遇到过这个问题？

浏览 1提问于2017-07-14得票数 0

1回答

为保修目的备份原始笔记本电脑的有效方法？

boot、partitioning、backup

我买了一台新惠普笔记本电脑。我将彻底消除Windows和安装Ubuntu (不是双重引导)。但是，如果我需要将笔记本电脑发回保修期，我希望能够将磁盘恢复到原来的状态(引导+分区+原始OS +原始软件+等等)。将原始1TB磁盘保存到网络位置的最佳方法是什么，因此我不需要> 1TB备份驱动器？此外，解决方案的适当恢复步骤是什么？ (注意:在第一次启动时，我将简单地引导到)。

浏览 0提问于2017-11-01得票数 0

回答已采纳

1回答

应用程序无法启动，因为kafka streams存储正在等待运行

java、spring-boot、apache-kafka、apache-kafka-streams

我有一个使用kafka streams的spring boot应用程序(kafka docker镜像：wurstmeister/kafka:2.12-2.1.1，kafka依赖：org.apache.kafka:kafka-streams:2.4.1)。在应用程序启动期间，我检查是否创建了主题my-topic，如果没有- application创建了它。在该应用程序创建KTable之后，如下所示： streamsBuilder.table("my-topic", Consumed.with(Serdes.String(), Serdes.String()), Materiali

浏览 1提问于2020-05-22得票数 0

1回答

有多个主题分区的Kafka流无法重新均衡错误

apache-kafka、apache-kafka-streams

当源主题分区计数= 1时工作正常。如果我将分区增加到任何大于1的值，我会看到下面的错误。既适用于低级，也适用于DSL API。有什么建议吗？可能会遗漏什么？ org.apache.kafka.streams.errors.StreamsException: stream-thread [StreamThread-1] Failed to rebalance at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:410) at org.apach

浏览 3提问于2017-02-19得票数 6

回答已采纳

1回答