kafka流中的处理器节点

Question

我正在Kafka流中的处理器节点上工作。对于一个简单的代码，我编写如下代码只是为了过滤用户ID，这是在kafka流中处理处理器节点的正确方法吗？

但是，下面的代码没有编译，抛出了一个错误：the method filter（谓词

KStreamBuilder builder = new KStreamBuilder();

builder.stream(topic)
    .filter(new Predicate <String, String>() {
        //@Override
        public boolean test(String key, String value) {
            Hashtable<Object, Object> message;
            // put you processor logic here
            return message.get("UserID").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);

有人能给我引路吗？




Answer 1

可能您正在使用另一个包中的Predicate类。您需要使用

import org.apache.kafka.streams.kstream.Predicate;

Answer 2

匿名用户

<代码>生成器。流（主题）返回KStream

如果您知道，实际类型是KStream

builder.<Sting,String>stream(topic)
       .filter(...)

要回答您关于“处理器节点”的问题：是的，添加一个过滤器（）将在内部添加一个处理器节点。注意，在DSL级别，通常不需要考虑处理器。

如果您想显式使用处理器，您可以使用处理器API而不是DSL。查看WordCount示例：https://github.com/apache/kafka/blob/trunk/streams/examples/src/main/java/org/apache/kafka/streams/examples/wordcount/WordCountProcessorDemo.java

注意，使用DSL，代码将在内部转换为处理器拓扑，即Kafka流的运行时模型。