我一直在检查Kafka流.我一直在测试以下代码用于Kafka流
I have been checking Kafka streams. I have been testing the below code for Kafka streams
生产者主题:(这是第一个生产者主题-发送以下json数据)
Producer topic: (this is the first producer topic - which sends the below json data)
KafkaProducer<String, String> producer = new KafkaProducer<>( properties); producer.send(new ProducerRecord<String,String>(topic, jsonobject.toString())); producer.close();JSON-主题的生产者:
JSON - Producer from topic:
{"UserID":"1","Address":"XXX","AccountNo":"234234","MemberName":"Stella","AccountType":"Savings"}流主题代码:(这是第二个流代码和主题)
Stream Topic code: (this is the second Streaming code and topic)
builder.<String,String>stream(topic) .filter(new Predicate <String, String>() { @Override public boolean test(String key, String value) { // put you processor logic here System.out.println("value : " + value); return value.substring(0).equals("1"); } }) .to(streamouttopic); final KafkaStreams streams = new KafkaStreams(builder, props); final CountDownLatch latch = new CountDownLatch(1); // attach shutdown handler to catch control-c Runtime.getRuntime().addShutdownHook(new Thread("streams-shutdown-hook") { @Override public void run() { streams.close(); latch.countDown(); } }); try { streams.start(); latch.await(); } catch (Throwable e) { System.exit(1); } System.exit(0);如果UserID值为"1",我想进行过滤,然后将该数据发送到目标流主题.
I want to filer if UserID value is "1", then send that data to destination streaming topic.
当我使用".filter"并打印System.out.println("value:" + value);时,执行时会引发以下错误.
When I use ".filter" and print System.out.println("value : " + value);, it throws the below error when executing.
Exception in thread "SampleStreamProducer-a6bb543e-bb92-48d0-8d9f-225046722d81-StreamThread-1" java.lang.ClassCastException: [B cannot be cast to java.lang.String如果我不使用".filter"并使用像这样的简单代码builder.stream(topic).to(streamouttopic);,则可以正常工作,但无需过滤.但是,我需要使用该过滤器.
If i don’t use ".filter" and use simple code like this, builder.stream(topic).to(streamouttopic); , it is working fine, but without filtering. But, I need to use that filter.
有人可以指导我进行修复吗?
Can someone guide me to fix it?
推荐答案默认情况下,Kafka Streams假定数据类型为<byte[],byte[]>,并且byte[]无法转换为String.
By default, Kafka Streams assumes data type <byte[],byte[]> and a byte[] cannot be cast to a String.
将主题读为KStream时,需要指定正确的Serdes:
You need to specify the correct Serdes when reading the topic as KStream:
builder.<String,String>stream(topic, Consumed.with(Serdes.String(), Serdes.String()) .filter(...)请查看示例并阅读文档:
Please check out the examples and read the docs:
- github/confluentinc/kafka-streams-examples
- kafka.apache/11/documentation/streams/
- github/confluentinc/kafka-streams-examples
- kafka.apache/11/documentation/streams/
更多推荐
过滤卡夫卡流
发布评论