配置kafka主题包含最近24小时的记录

文章/答案/技术大牛

发布

1回答

apache-kafka

我希望配置一个Kafka主题，这样每条记录都有24小时的TTL，从它第一次写入到从主题中删除使用这个页面：https://kafka.apache.org/documentation/#brokerconfigs我看过一些属性，但它们都引用了日志的保留，而不是单个记录。可以在主题上配置基于记录的TTL吗？

浏览 18提问于2020-07-27得票数 0

回答已采纳

1回答

Debezium将1GB数据库转换为100 1GB主题存储。

json、postgresql、apache-kafka、storage、debezium

我在容器中有Debezium，捕获PostgeSQL数据库记录的所有更改。我知道从转换到JSON会有一个开销，但是两者之间的差别要大很多倍。有没有什么可以被配置来减少kafka主题存储的东西？

浏览 16提问于2022-10-31得票数 1

回答已采纳

1回答

Kafka源JDBC连接器-微批处理

jdbc、apache-kafka、apache-kafka-connect

作为POC的一部分，我有一个任务，它基本上读取MySQL表的所有内容，并将记录转储到一个Kafka主题中。该表仅包含2条记录&我的batch.max.rows值为2。当任务以"bulk“模式运行时，我在kafka主题中看到了2条单独的JSON记录。如何配置连接器以插入包含这2条记录的JSON

浏览 0提问于2021-01-06得票数 1

回答已采纳

1回答

如何存储主题上的数据，这些数据将通过生产者作为消息传递，以便可以加载到相应列的数据库中

apache-kafka、cassandra

到目前为止，我使用kafka spring在this blog上生成了一条主题消息。我要传递的消息只是名称，现在我想流过这个主题，并添加增量值作为ID和名称，并将其存储在OutputTopic上，现在，我想将数据存储到cassandra。我在cassandra中的表结构如下： CREATE TABLE emp( emp_name text, ) 输出主题上的数据应该是哪种格式

浏览 12提问于2021-08-25得票数 1

回答已采纳

1回答

Kafka版本:3.0.1-Kafka管理客户端反复创建-内存泄漏

spring-boot、apache-kafka、out-of-memory、spring-kafka、spring-cloud-stream

我们有一个跳转应用程序，它从一个主题消耗，并生成记录到多个主题。 org.apache.kafka:kafka-clients:3.0.1 org.springframework.cloud:spring-cloud-stream:spring-cloud-stream-binder-kafka</

浏览 17提问于2022-08-18得票数 0

1回答

如何在Spring Cloud Stream Kafka* Binder中设置死信队列的保留时间？*

apache-kafka、spring-cloud-stream、dead-letter、spring-cloud-stream-binder-kafka、retention

我有一个使用Spring Cloud Stream Kafka的应用程序。对于用户定义的主题，我可以通过下面提到的配置从指定的主题中删除记录。但此配置不适用于DLQ主题。例如，在下面的配置中，我在绑定器级别配置了保留时间。因此，在绑定级别下定义的我的生产者主题(学生主题)配置

浏览 51提问于2021-03-12得票数 1

回答已采纳

1回答

Kafka使用者在不查询DB每条消息的情况下摄取数据

database、apache-kafka、kafka-consumer-api

我们有一个卡夫卡消费者服务，将数据摄入到我们的数据库中。每当我们从主题中接收到消息时，我们将编写一个insert语句将此消息插入到DB中。我们使用DB连接池来处理插入，到目前为止还不错。目前，我们需要添加一个过滤器，以便只从Kafka选择相关消息并执行插入操作。我有两个选择来做这件事。Pros这不是

浏览 3提问于2022-11-02得票数 0

1回答

Kafka avró无法定位主题

java、apache-kafka、avro、confluent-schema-registry

我正在生成一个Kafka记录，使用在schema Registry中注册的Avro模式。该主题已注册，因为当我点击时{version: 2,schema: "{"type":"record","name"::"string","doc":"The string is a unicode characte

浏览 1提问于2020-07-19得票数 0

1回答

卡夫卡物化观点

apache-kafka、apache-kafka-streams、materialized-views

据我所知，卡夫卡默认情况下将在主题中保留记录7天，然后删除它们。但是卡夫卡物化的观点又如何呢，卡夫卡将把数据保存多久(不定式还是有限的时间)？同样，卡夫卡是否复制了集群上的物化视图？

浏览 1提问于2018-08-26得票数 1

回答已采纳

1回答

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

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

1回答

Kafka接收器连接-如何为每个主题(表)创建一个任务

apache-kafka、apache-kafka-connect、confluent-platform

我们已经为我们的产品实现了一个名为Ampool ADS的kafka接收器连接器，它将来自kafka主题的数据摄取到相应的ampool表中。主题和表通过它们的名称映射。我需要将单个主题(从topic ->表中摄取)处理到一个专用的接收器任务中。因此，例如，如果我的配置包含3个不同的主题(topic1、top

浏览 2提问于2017-04-24得票数 0

1回答

卡夫卡连接HDFS配置设置？

apache-kafka、apache-kafka-connect、confluent-platform

我想配置Kafka Connect，以便将尽可能多的记录写入HDFS (我需要每天编写大约100 B记录，每条记录大约为1.2Kb)。flush.size=305000有215个分区的Kafka主题。对于这个设置，我有一个例外： flush.s

浏览 0提问于2018-08-29得票数 0

回答已采纳

1回答

卡夫卡流中信息的无序处理

apache-kafka、apache-kafka-streams

最近，我看到了关于Apache Kafka文档中有关处理Kafka流中的无序消息的文章。在主题分区中，记录的时间戳可能不会随偏移量单调增加。由于Kafka流总是试图处理主题内的记录--按照偏移顺序进行分区，因此它可能导致具有较大时间戳(但偏移量较小)的记录比同一主题

浏览 0提问于2019-02-04得票数 0

1回答

Ignite Source Kafka* Connector是否提供数据的主题分区？*

apache-kafka、ignite

但由于高负载和只有一个分区，我们有时会得到如此多的延迟，这可能不是我们未来的最佳解决方案。因此，我们计划有多个消费者组，但由于我们将只有一个分区，这将导致一致性问题，因为我们需要队列排序。我们可以通过IgniteSourceConnector为主题配置多个分区吗

浏览 21提问于2021-07-21得票数 0

2回答

卡夫卡流-在源不是卡夫卡主题的情况下，我可以使用卡夫卡流处理吗？

apache-kafka-streams

基于的内容(回调类型)，我必须将其放入相应的传出主题(例如账单-n-cdr，dr-cdr，.)。我在互联网博客上看到的所有例子，用阅读源主题和其他目的地主题的上下文来解释热气腾腾的应用程序。那么，这是否可以从非主题源(例

浏览 4提问于2022-06-10得票数 0

1回答

KStream-KStream联接与KStream-KTable联接性能

apache-kafka、apache-kafka-streams

我有一个Kafka流应用程序，它从几个主题中获取数据，并将数据加入到另一个主题中。Kafka配置：Kafka Topics - 15 partitions and 3 replication factor. 在性能和其他因素方面哪个更好

浏览 2提问于2017-09-14得票数 0

2回答

使用Flink动态创建和更改Kafka主题

apache-kafka、apache-flink

我使用Flink从不同的Kafka主题读取和写入数据。具体来说，我使用的是FlinkKafkaConsumer和FlinkKafkaProducer。我想知道是否有可能根据我程序中的逻辑或记录本身的内容来更改我正在阅读和写入的Kafka主题。例如，如果读取了包含新字段的记录，我希望创建一个新主题，并开始将包含该字段的<

浏览 5提问于2018-08-18得票数 2

1回答

在Kafka连接器中，如何获得我的Kafka* Connect当前使用的引导服务器地址？*

apache-kafka、apache-kafka-connect

我的反序列化器是JSONConverter。但是，当有人向我的连接器的主题发送错误的JSON数据时，我希望省略此记录，并将此记录发送到我公司的特定主题。我的困惑是:我找不到任何API来获取我的连接的bootstrap.servers。(我知道它在confluent的etc目录中，但编写"connect-distributed.properties“目录<

浏览 0提问于2018-11-20得票数 1

回答已采纳

1回答

如何在librdkafka中设置复制因子？

apache-kafka

我正在使用librdkafka在C++ kafka message producer中开发。有没有办法创建不同于默认复制因子的自定义复制因子的topic？CONFIGURATION.md没有明确提到任何参数，但Kafka工具允许这样做。

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

回答已采纳

1回答

kSQLDB声明密钥字段在流或表上创建随机密钥

apache-kafka、ksqldb

我正在使用kSQLDB并从一个主题创建一个流，当我显式声明一个KEY时，我会得到一个不属于我的键的随机值。如果删除KEY关键字，我需要的值实际上是在Id列中指定的。 -- omitdeclaring id as the key

浏览 3提问于2021-10-20得票数 0

点击加载更多