首页 文章

Kafka Stream根据json消息中的时间戳键对消息进行排序

提问于
浏览
0

我正在使用JSON消息发布Kafka,例如:

"UserID":111,"UpdateTime":06-13-2018 12:13:43.200Z,"Comments":2,"Like":10
"UserID":111,"UpdateTime":06-13-2018 12:13:40.200Z,"Comments":0,"Like":6
"UserID":222,"UpdateTime":06-13-2018 12:13:43.200Z,"Comments":1,"Like":10
"UserID":111,"UpdateTime":06-13-2018 12:13:44.600Z,"Comments":3,"Like":12

我想使用Kafka Streams在10秒的时间窗口中基于 UpdateTime 对消息进行排序,并在另一个Kafka主题中推回已排序的消息 . 我创建了一个流,它从输入主题读取数据,然后我在 groupByKey() 之后创建 TimeWindowedKStream ,其中UserID是消息中的键(虽然它不需要 groupByKey 然后排序,但我无法直接获得 WindowedBy ) . 但我无法在10秒窗口内根据_422695进一步对消息进行排序 . 我的源代码是:

public static void main(String[] args) throws Exception {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-sorting");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "broker");
        props.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 0);
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> source = builder.stream("UnsortedMessages");
        TimeWindowedKStream<String, String> countss = source.groupByKey().windowedBy(TimeWindows.of(10000L)
                 .until(10000L));
        /*
        SORTING CODE
            */
        outputMessage.toStream().to("SortedMessages", Produced.with(Serdes.String(), Serdes.Long()));
        final KafkaStreams streams = new KafkaStreams(builder.build(), props);
        final CountDownLatch latch = new CountDownLatch(1);
        // attach shutdown handler to catch control-c
        Runtime.getRuntime().addShutdownHook(new Thread("streams-sorting-shutdown-hook") {
            @Override
            public void run() {
                streams.close();
                latch.countDown();
            }
        });
        try {
            streams.start();
            latch.await();
        } catch (Throwable e) {
            System.exit(1);
        }
        System.exit(0);
    }

提前谢谢了 .

1 回答

  • 0

    如果要对忽略密钥的消息进行排序,则仅根据分区执行此操作,并且仅当输入主题与输出主题具有相同数量的分区时才有意义 . 对于这种情况,您应该提取分区号并将其用作消息密钥(cf:https://docs.confluent.io/current/streams/faq.html#accessing-record-metadata-such-as-topic-partition-and-offset-information

    对于排序,它更棘手 . 请注意,Kafka Streams遵循"continuous output"模型,并使用DSL为每个输入记录发出更新 . 因此,使用Processor API可能更好 . 您将使用带有附加商店的 Processor 并将记录放入商店 . 作为内存中的结构,您可以保留已排序的记录列表 . 当时间推进时,您可以发出"finished"窗口并从商店中删除相应的记录 .

    我不认为你可以使用DSL来构建它 .

相关问题