处理二进制图像数据的Java Kafka使用者

文章/答案/技术大牛

发布

1回答

将processing.guarentee设置为exactly_once会在python使用者kafka创建反序列化错误。

python、apache-kafka、confluent-kafka-python

我有java Kafka流处理应用程序和python应用程序。Java应用程序生成数据，python使用者使用它。当processing.guarantee设置为exactly_once时，python使用者无法反序列化数据。反序列化失败。我尝试了一个java使用者，而java使用者正在成功地读取数据。在两种情况下，即使是在反序列化之前在python使用

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

回答已采纳

1回答

java、apache-kafka

我刚接触Kafka，目前需要提取二进制图像数据。我目前使用的是KafkaAvroDeserializer。如何提取"bytes“原始字节数据？

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

2回答

如何查询kafka获得主题中给定键的最新消息的最新偏移量/分区/timestamp

java、go、apache-kafka、kafka-consumer-api、sarama

我如何查询kafka并获得最新的、最近的消息的时间戳/偏移量/分区，以及主题和消息键？我们的应用程序有高丽和爪哇。然而，任何解决办法都是受欢迎的。提前谢谢。

浏览 8提问于2022-06-17得票数 1

1回答

我启动了我的存贮器和kafkaServer，并能够从生产者发送消息给消费者，但我不知道如何处理一个图像，并将它从生产者发送到消费者？./bin/kafka-server-start.sh config/server.properties./bin/kafka-topics.sh --create --zookeeper localhost:2181 -- replication-factor 1 --partitions

浏览 1提问于2019-04-09得票数 1

3回答

Kafka消费者没有通过Java读取主题数据

java、apache-kafka、kafka-consumer-api

我向Kafka主题发送输入的JSON数据。通过使用下面的命令，我可以在kafka使用者中看到相同的JSON数据。 .\bin\windows\kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic mytopic --from-beginningCase1:，但是当我试图读取kafka使用者( java )的<

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

1回答

错误处理程序抛出一个异常: Kafka使用者CommitFailedException

spring-boot、resttemplate、spring-kafka

无法完成偏移提交，因为使用者不是用于自动分区分配的活动组的一部分；很可能使用者被踢出了组。卡夫卡是工作的，如果它是消费消息和消费的消息将发送到一个服务通过restTemplate，但不知怎的restTemplate是失败的连接，但我得到以下的卡夫卡错误。当restTemplate失败时，卡夫卡消费的问题是什么？$ListenerConsumer -错误处理程序抛出异常org.apache.kafka

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

回答已采纳

1回答

用kafka从甲骨文中传输数据

oracle、apache-kafka、data-stream

我从kafka开始，我需要控制特定Oracle表中的插入，现在通过kafka发送新记录。我无法控制数据库，所以原则上，Debizium被排除在外。我该怎么做？不用触发器。我让一个生产者使用eclipse中的java程序从Oracle读取数据，但是，这将对数据库发出不断的请求。我使用java与使用者一起模拟ETL。 PS:我和Windows一起工作，但那是次要的。

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

1回答

避免Kafka数据中心之间的重复数据处理

apache-kafka

在主动-主动Kafka设计中，当数据从DC1复制到DC2 (具有相同或不同区域)时。如何确保DC2使用者不会处理DC1使用者已经处理过数据？应该做什么Kafka级别的配置？

浏览 11提问于2021-02-10得票数 0

2回答

将Dstream中的分区数目增加到比直接方法中的Kafka分区大

apache-spark、apache-kafka、spark-streaming、rdd、dstream

它们是32个Kafka分区和32个使用者(按照Direct )。但是32位消费者的数据处理速度比Kafka慢(1.5倍)，这在Kafka中造成了大量的数据积压。我希望增加每个使用者接收的Dstream分区的数量。我希望解决方案能够在消费者身上增加分区，而不是在Kafka中增加分区。

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

回答已采纳

1回答

在java代码中检查Kafka使用者组的活泼性

java、apache-kafka

我已经为Kafka使用者编写了一个java代码，它使用executor服务，它将运行24/7，使用数据并处理它。但是在某些情况下，例如由于Kafka协调器代理节点崩溃，作业将失败--即消费者组将死亡，我将无法知道它，而java jar将运行但什么也不做。INFO [2018-11-07 03:25:58,425] [pool-1-thread-1] (AbstractCoordinator.java:coordinatorDead) - Ma

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

回答已采纳

1回答

如何在不分割topic的情况下实现多处理，消费kafka消息？

apache-kafka、kafka-consumer-api

我的主题只有一个分区，但我需要实现多进程。我有大量异步生成的消息，我希望异步读取所有这些消息并提交每条消息。

浏览 2提问于2018-10-03得票数 4

2回答

试图通过Logstash传递二进制文件

amazon-s3、apache-kafka、logstash

一些进程正在生成我的Kafka二进制文件(来自Java，它是字节数组)。input { bootstrap_servers => "my-broker:9092" partition_assignment_strategy=> "org.ap

浏览 5提问于2020-06-02得票数 3

回答已采纳

2回答

使用Kafka、Storm和HBase进行实时处理

hbase、apache-kafka、apache-storm

为我的无知道歉。我们正在建立一个实时数据库，捕获基于网站上的成员活动的记录。发出这些信息的应用程序是IBM MQ。我们计划将Apache Kafka设置为使用来自IBM MQ的这些消息，并将其存储在HBase for Analytics中。Kafka有没有可能直接将这些信息写入HBase，或者我们应该在Kafka和HBase之间设置一个像Apache Storm这样的实时流引擎？这就是Kafka

浏览 1提问于2015-06-25得票数 1

1回答

卡夫卡使用者错误:运行使用者时的错误未知错误：(kafka.tools.ConsoleConsumer)

apache-kafka

在我的例子中，我让Kafka二进制kafka_2.11-1.0.0同时安装在服务器端和客户端，但是在创建完主题之后，我的使用者在我使用--引导服务器而不是--动物园管理员时没有工作。随着警告的到来，我改变了。请您更新一下为什么使用者不使用预期的用户，而是使用旧的调用使用者的方式。

浏览 0提问于2018-04-12得票数 6

1回答

如何从原始缓冲区对象Kafka* java中获取字符串消息*

java、node.js、apache-kafka、node-kafka

当卡夫卡的消费者消费了卡夫卡的信息时，看起来- {"type":"Buffer","data":[91,13 .....]但我需要在java Kafka消费中消费。我的问题是如何使用由nodejs生产者生成的java字符串格式的消息。

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

1回答

Apache处理流水线-应用状态、错误处理和容错？

java、apache-beam、amazon-kinesis、apache-beam-io

我熟悉Kafka的概念，了解它如何处理消费者的偏移量/状态，并具有实现apache / experience处理的经验。在阅读了这些文档之后，我成功地创建了一个工作波束管道，使用KinesisIO Java侦听AWSKinsis数据流来转换和打印消息。但是，想知道在apache w.r.t中如何处理以下区域的引用实现或指针。KinesisIO - 如何在Kinesis中唯一地标识消费者应用程序(类

浏览 3提问于2020-08-21得票数 0

1回答

使用avro的Kafka控制台生产商

spring-boot、apache-kafka、confluent-platform

我正试图使用apache-kafka二进制文件中的kafka-控制台-生产者来生成消息，并在春季启动时从消费者设置中消费。使用者使用avro架构。当以json格式生成消息时，我的使用者抛出异常--“无法序列化”。我找到了一个解决方案来使用“7.1”，其中有kafka-avro-console-producer。它支持avro，但它是企业版。是否有一种方法可以使用apache-kafka与kafka控制台生产者一起使

浏览 6提问于2022-04-26得票数 0

回答已采纳

1回答

是否自己创建了一个消费者组(因为它不关心应用程序中给出的group.id属性)

apache-spark、apache-kafka、spark-streaming、spark-streaming-kafka

对于第一批，驱动程序中的流上下文连接到Kafka并获取startOffset和endOffset。然后，启动具有这些开始和结束偏移范围的火花作业，以便执行者从Kafka获取记录。我的问题从这里开始。当第二批的时间，流上下文连接到卡夫卡的开始和结束偏移范围。如果没有允许存储最后一次提交偏移值的使用者组(因为直接流没有考虑到group.id)，那么Kafka如何能够提供这些范围呢？

浏览 4提问于2020-02-27得票数 0

回答已采纳

2回答

如何使用Apache进行内容过滤？

java、apache-kafka

我需要做的是过滤由生产者产生的消息，根据它们的前缀。例如，如果消息以'a‘前缀开头，那么只有第一个使用者必须接受它。如果它以'b‘前缀开头，那么只有第二个消费者必须接受它。我做了很多搜索，我发现的是过滤来自一个主题的消息，然后在过滤后将它们发送到不同的主题。但如前所述，我需要对一个主题进行过滤。我怎么能在卡夫卡做到这一点？

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

回答已采纳

7回答

卡夫卡消费者违约组Id

apache-kafka、kafka-consumer-api

我正在使用Apache及其Java客户端，我看到在属于同一组(即共享同一个组id)的不同Kafka使用者之间，消息是负载均衡的。编辑:谢谢你的回答。你是对的:如果一个人没有设定消费者

浏览 17提问于2017-03-30得票数 38

点击加载更多

将processing.guarentee设置为exactly_once会在python使用者kafka创建反序列化错误。