过滤 Kafka 流

问题描述

我一直在检查 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 代码:(这是第二个Streaming 代码和主题)

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[] 不能转换为 String.

By default, Kafka Streams assumes data type <byte[],byte[]> and a byte[] cannot be cast to a String.

阅读题目时需要指定正确的Serdes为KStream:

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:

• https://github.com/confluentinc/kafka-streams-examples
• https://kafka.apache.org/11/documentation/streams/

这篇关于过滤 Kafka 流的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！