如何在 Kafka 流中打印 TimeWindowedKStream 和 KTable？答案

【问题标题】：How to print TimeWindowedKStream and KTable in Kafka streams?如何在 Kafka 流中打印 TimeWindowedKStream 和 KTable？
【发布时间】：2020-08-06 22:57:01
【问题描述】：

我们有一个 Kafka 进程，它将主题作为输入并将定时窗口写入输出主题。正在使用以下代码。我想打印 TimeWindowedKStream(groupedStream) 和 KTable(aggregatedTable) 并查看输出以进行一些调试..

    String intopic = input_topic;
    Long window = 60;
    String outtopic = output_topic;
    
    final Serde<String> stringSerde = Serdes.String();

    Properties property = new Properties();
    property.put("bootstrap.servers", "127.0.0.1:9092");
    property.put("group.id", "test-consumer-group");
    property.put("application.id", "sliding-window-min-bar");
    property.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, stringSerde.getClass().getName());
    property.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, stringSerde.getClass().getName());

    Duration windowSizeMs = Duration.ofMinutes(window);

    StreamsBuilder builder = new StreamsBuilder();
    
    System.out.println(intopic);

    KStream<String, String> equitybar = builder.stream(intopic, Consumed.with(stringSerde, stringSerde));
    
    System.out.println(equitybar);
    
    equitybar.print(Printed.toSysOut());

    // convert string of csv to a double on the mean value
    KStream<String, String> transformedbar = equitybar
            .map((key, value) -> KeyValue.pair(key, value.substring(1,value.length()-2).split(",")[2]));

    System.out.println(transformedbar);

    transformedbar.print(Printed.toSysOut());
    
    // group by equity and sliding window
    
    System.out.println(windowSizeMs);
    System.out.println(TimeWindows.of(windowSizeMs).advanceBy(advanceMs));
    
    TimeWindowedKStream<String, String> groupedStream = transformedbar.groupByKey().windowedBy(TimeWindows.of(windowSizeMs).advanceBy(advanceMs));

    System.out.println(groupedStream);
    KTable<Windowed<String>, String> aggregatedTable = groupedStream.aggregate(
            () -> "|",
            (aggKey, newValue, aggValue) -> aggValue + newValue.trim() + "|")      ;

我尝试使用用于 Kafka 流的打印命令来打印它 - groupedStream.print(Printed.toSysOut()); - 但它似乎不起作用。

谢谢。

【问题讨论】：

标签： apache-kafka apache-kafka-streams

【解决方案1】：

KGroupedStream 和 TimeWindowedKStream 是“只是”帮助类，允许 DSL 向链式运算符提供流畅的 API，而不会在单个类上出现过多的重载。

在 DSL 中，只有两个主要抽象，KStream 和 KTable，它们是实际的一流数据容器。因此，您想要做的事情是不可能的。

【讨论】：