Kafka with Java:如何重新读取数据

文章/答案/技术大牛

发布

1回答

java、apache-kafka、commit、kafka-consumer-api

我对kafka API有以下问题。我用以下命令设置我的客户：然后我想从Kafka中读取数据，从这些数据中更新数据库。所以我想将DB事务应用到kafka，i.a。如果我的数据库事务是OK的，那么移动kafka指针，但如果它失败了，那么从相同的位置重试。在我的代码中， <code>A2<

浏览 31提问于2018-12-21得票数 1

回答已采纳

1回答

如何用火花结构流管理卡夫卡的偏移量

scala、spark-structured-streaming

我有一个火花结构化的流媒体工作，它需要读取kafka主题中的数据并进行一些聚合。作业需要每天重新启动，但是当它重新启动时，如果我设置了startingOffsets="latest"，就会丢失在重新启动之间的数据。如果我设置了startingOffsets="earliest"，那么作业将从主题读取所有数据，而不是从上一个流作业离开的位置读取数据。有人能帮助我

浏览 1提问于2019-05-15得票数 1

回答已采纳

1回答

如何在Spring Boot Camel中使用onException()处理异常

java、spring-boot、apache-kafka、apache-camel、spring-camel

Spring boot Apache Camel-Java DSL应用程序从Kafka主题读取消息。Component from("kafkabrokers=localhost:9092") .log("Message received from Kafka: ${body}&

浏览 66提问于2020-11-29得票数 0

2回答

Apache Beam over Apache Kafka流处理

apache-kafka、apache-beam、apache-kafka-streams、stream-processing

Apache Beam和Apache Kafka在流处理方面有什么不同？我也在努力掌握技术和编程上的差异。请通过您的经验报告来帮助我理解。

浏览 1提问于2018-06-15得票数 10

1回答

将Kafka 0.8.2.1集群中的数据镜像到Kafka 2.2.0集群

apache-kafka、replication

我想使用Apache Spark Structured Streaming和Kafka，Spark Structured Streaming支持Kafka 0.10及更高版本，我的Kafka集群使用kafka我想将当前kafka 0.8.2.1集群中的一些主题复制到基于2.2.0的新Kafka集群中。为此，我尝试在Kafka 2.2.0集群上使用kafka-console-consumer来监听来自kafka集群0.8.2.1的消息，

浏览 17提问于2019-05-23得票数 1

1回答

在storm 1.0.2中，kafka-spout每次重新启动拓扑都会重复消耗相同的数据

apache-kafka、apache-storm

(我使用AVRO类)在更改版本之前，请正确使用下一个数据，即使重新启动拓扑也是如此。数

浏览 0提问于2016-11-18得票数 1

2回答

Spark Streaming: Kafka重启后输入速率不正常

apache-kafka、spark-streaming

我目前正在spark上运行一个流媒体应用程序，它使用新的直接方法(没有接收器)消费一个简单的kafka主题。由于我们在Kafka集群上遇到了一些问题，即使在Kafka上恢复正常后，流媒体应用程序也遇到了一些问题:一些作业正在处理几乎空的rdd，而另一些作业正在处理两倍大小的rdd。我在Kafka和Spark两边都没有发现任何堆栈或线索。有人能告诉我如何解决这类问题吗？或者至少到哪里去看一看，以了解发生了什么？

浏览 0提问于2016-05-18得票数 0

3回答

Kafka消费者(0.8.2.2)是否可以批量读取消息

java、apache-kafka、kafka-consumer-api

根据我的理解，Kafka消费者顺序地从分配的分区读取消息...我们计划让多个Kafka客户(Java)拥有相同的组I'd ..so，如果它按顺序从指定的分区读取数据，那么我们如何实现高吞吐量..i.e。在我们的例子中，消费者只有在消息成功处理后才必须提交偏移量，..else消息将被重新处理...还有更好的解决方案吗？

浏览 0提问于2016-02-25得票数 5

2回答

Kafka Streams - java版本依赖关系

apache-kafka-streams

我正在使用Java1.6运行我的应用程序，并计划将一些方面与Kafka集成。我正在探索是否应该使用Kafka Consumer (使用poll())或使用KafkaStreams API。我们需要Java 1.7 / 1.8来使用Stream API吗？

浏览 3提问于2018-12-20得票数 0

1回答

与Redis和Prometheus集成

redis、apache-kafka、prometheus

我想从kafka主题中读取指标，并使用http端点公开指标，以便普罗米修斯可以抓取数据。如何使用Redis Cache公开自定义指标？我需要听取kafka主题，并使用redis公开指标。我们使用java来读取kafka topic中的指标。

浏览 21提问于2019-08-03得票数 0

回答已采纳

1回答

Kafka stream应用程序可以从Kinesis Stream中读取吗？

java、apache-kafka、apache-kafka-streams、amazon-kinesis

我需要写一个java应用程序，从Kinesis流读取和存储数据到Kafka主题。我不能使用连接器，我必须使用Kafka Streams，因为其中涉及到很多逻辑。Kafka Stream是否可以直接从Kinesis流读取，或者我必须先将数据存储到Kafka topic才能被Kafka Stream读取？我读了很多关于Kafka Streams的文章，除了Kafka主

浏览 18提问于2019-07-29得票数 1

1回答

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

apache-kafka、apache-kafka-streams

我开始研究如何重新划分压缩的主题，方法是使用一个groupBy()，后面跟着一个()，但是当它开始读取创建的重新分区主题时，它开始抛出一个带有消息的：重新分区是由KafkaStreams自动创建和填充的，所以这看起来很奇怪，它会写一个无效的记录。(FailOnInvalidTimestamp.java:48) at org.apache.ka

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

回答已采纳

1回答

如何读取按特定字段过滤的Kafka表？

java、apache-kafka

ROWTIME":123456, "id":1, "person_id":"1234"在这个kafka表中，我有很多具有不同provider_id的不同数据。我需要从这个kafka表中检索所有不同的person_ids，其中provider=3。我是卡夫卡的新手，在这里找到了这个方法：但我不确定

浏览 0提问于2021-04-28得票数 0

1回答

Debezium SQL Server源连接器在连接器重新启动后跳过记录

sql、sql-server、apache-kafka-connect、debezium

我使用Debezium CDC SQL Server源连接器从SQL server获取数据并将其放入Kafka中，我使用的配置是connector.class=io.debezium.connector.sqlserver.SqlServerConnectordatabase.dbname=demodatabase.history.kafka.bootstrap.servers=localhost:9092database.history.kafka.

浏览 0提问于2020-06-26得票数 0

1回答

卡夫卡消费者再平衡及其影响

apache-kafka、kafka-consumer-api

我是Kafka的新手，我正在尝试用Java和Go (使用Confluent/ Kafka -Go)设计一个包装库，以便在内部使用Kafka。对于我的用例，CommitSync是关键的一步，我们应该只在正确提交旧的之后才进行读取。重复处理不是一个大问题，我们的客户端服务已经足够强大了。但数据丢失是一个主要问题，不应该发生。当网络恢复时，Kafka消费者会自动处理并激活吗?还是我们必须重新初始化它们？如果他们活着回来，他们会从离开的地方继续工作吗

浏览 2提问于2019-03-18得票数 1

1回答

Spring Kafka* JDBC连接器兼容性*

spring-boot、spring-kafka、apache-kafka-connect、confluent-platform

Kafka JDBC connect兼容Spring-Kafka库吗？我确实关注了，但仍然有一些困惑。

浏览 4提问于2020-07-04得票数 0

1回答

Kafka流:如何转换数据类型？

java、apache-kafka、kafka-consumer-api、apache-kafka-streams

我使用Kafka的Streams和拓扑构建器。我想知道如何拥有一个处理器，它可以将一种数据类型转换为另一种数据类型，这样，管道中的下一个处理器就可以使用它了。

浏览 1提问于2017-10-30得票数 2

2回答

卡夫卡中的UTF8-mb4

java、apache-kafka、utf-8、utf8mb4

我在mongo中有一个UTF8-mb4字符，还有一个java提取器，它从mongo提取数据并放入kafka。当数据转到Kafka时，特殊的字符被替换为\u.Sample text:- "\uDBFF\uDC15COMPANY" 我有另一个Java程序，它从一个kafka主题中读取，并在经过一些操作之后将它放到另一个kafka主题中。当数据从实际主题中读取时\u..。被解码为一个实

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

1回答

等效于python中的objectmapper

java、python、apache-kafka

我想用python序列化我的数据，以便将数据作为一个生产者发送给Kafka。:-) 还有一个带有java的客户端，它将使用java和kafka.common.serialization反序列化消息。换句话说，我应该如何使用python序列化数据，以便从使用者端(Java)读取数据？

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

1回答

我的生产者(SQL Server)打开了，Kafka关闭了。如何重新处理发往kafka的消息？

apache-kafka、apache-kafka-connect

我正在使用Debezium SQL Server Source connector通过CDC将数据推送到Kafka。SQL Server表中的数据是连续插入的。当Kafka宕机时，数据还会插入到该表中吗？如果是这样，当我的Kafka打开时，如何获得这些记录？

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

回答已采纳

点击加载更多