使用kafka-streams对json输入流进行条件排序

Question

我是开发kafka-streams应用程序的新手。我的流处理器用于根据输入json消息中的用户键值对json消息进行排序。

Message 1: {"UserID": "1", "Score":"123", "meta":"qwert"}
Message 2: {"UserID": "5", "Score":"780", "meta":"mnbvs"}
Message 3: {"UserID": "2", "Score":"0", "meta":"fghjk"}

我读到这里动态连接一个Kafka输入流到多个输出流，没有动态解决方案。

在我的用例中，我知道对输入流排序所需的用户键和输出主题。因此，我编写了针对每个用户的单独的处理器应用程序，其中每个处理器应用程序匹配不同的用户ID。

所有不同的流处理器应用程序都从kafka中的相同json输入主题读取，但每个应用程序都只在满足预设用户条件的情况下将消息写入特定用户的输出主题。

public class SwitchStream extends AbstractProcessor<String, String> {
        @Override
        public void process(String key, String value) {
            HashMap<String, String> message = new HashMap<>();
            ObjectMapper mapper = new ObjectMapper();
            try {
                message = mapper.readValue(value, HashMap.class);
            } catch (IOException e){}

            // User condition UserID = 1
            if(message.get("UserID").equals("1")) {
                context().forward(key, value);
                context().commit();
            }
        }

        public static void main(String[] args) throws Exception {
            Properties props = new Properties();
            props.put(StreamsConfig.APPLICATION_ID_CONFIG, "sort-stream-processor");
            props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
            props.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
            props.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

            props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
            TopologyBuilder builder = new TopologyBuilder();
            builder.addSource("Source", "INPUT_TOPIC");
            builder.addProcessor("Process", SwitchStream::new, "Source");
            builder.addSink("Sink", "OUTPUT_TOPIC", "Process");

            KafkaStreams streams = new KafkaStreams(builder, props);
            streams.start(); 
       }
}

问题1：如果使用低级处理器API，那么使用高级流DSL是否可以轻松实现相同的功能？（我承认我发现更难理解和跟随其他在线的高级流DSL示例）

问题2:input json主题以20k-25k EPS的高速率获取输入。我的处理器应用程序似乎无法跟上这个输入流的步伐。我曾尝试部署每个流程的多个实例，但结果与我希望的结果相去甚远。理想情况下，每个处理器实例应该能够处理3-5K EP。

有没有一种方法来改进我的处理器逻辑或编写相同的处理器逻辑使用高级流DSL？那会有什么不同吗？

Answer 1

您可以在高级DSL中通过filter()（您有效地实现了一个筛选器，因为您只返回一个消息，如果它是userid==1)。您可以通过使用kstream#branch()（有关更多详细信息，请参阅文档:http://docs.confluent.io/current/streams/developer-guide.html#Stateless-transformations)来概括这种筛选模式。也请阅读JavaDocs：http://kafka.apache.org/0102/javadoc/index.html?org/apache/kafka/streams

KStreamBuilder builder = new KStreamBuilder();
builder.stream("INPUT_TOPIC")
       .filter(new Predicate() {
           @Overwrite
           boolean test(String key, String value) {
               // put you processor logic here
               return message.get("UserID").equals("1")
           }
        })
       .to("OUTPUT_TOPIC");

关于表演。单个实例应该能够处理10k+记录。如果没有任何进一步的信息，很难判断问题可能是什么。我建议在Kafka用户列表中询问（见http://Kafka.apache.org/contact）