将 Kafka 流输入打印到控制台？

Posted 2023-02-15

【中文标题】将 Kafka 流输入打印到控制台？

【英文标题】：Print Kafka Stream Input out to console?

【发布时间】：2017-01-12 15:55:35

【问题描述】：

我一直在查看大量 Kafka 文档以了解我正在开发的 Java 应用程序。我已经尝试过使用 Java 8 中引入的 lambda 语法，但我在这方面有点粗略，并且对它应该是我目前使用的不太有信心。

我有一个 Kafka/Zookeeper 服务运行没有任何问题，我想做的是编写一个基于输入的小示例程序将其写出来，但不做字数统计，因为有这么多的例子已经。

至于示例数据，我将得到以下结构的字符串：

示例数据

This a sample string containing some keywords such as GPS, GEO and maybe a little bit of ACC.

问题

我希望能够提取 3 个字母的关键字并使用 System.out.println 打印它们。如何获取包含输入的字符串变量？我知道如何应用正则表达式，甚至只是通过字符串搜索来获取关键字。

代码

public static void main(String[] args) 
    Properties props = new Properties();
    props.put(StreamsConfig.APPLICATION_ID_CONFIG, "app_id");
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "0:0:0:0:0:0:0:1:9092");
    props.put(StreamsConfig.ZOOKEEPER_CONNECT_CONFIG, "0:0:0:0:0:0:0:1:2181");
    props.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
    props.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());

    final Serde<String> stringSerde = Serdes.String();

    KStreamBuilder builder = new KStreamBuilder();

    KStream<String, String> source = builder.stream(stringSerde, stringSerde, "in-stream");

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

    //How do I assign the input from in-stream to the following variable?
    String variable = ?

我的 zookeeper、kafka、producer 和 consumer 都运行在同一个主题上，所以我希望基本上看到相同的 String 出现在所有实例（producer、consumer 和 stream）上。

【参考方案1】：

如果您使用 Kafka Streams，您需要在数据流上应用函数/运算符。在您的情况下，您创建了一个 KStream 对象，因此，您希望将运算符应用于 source。

根据您想要执行的操作，有一些运算符可以独立地将函数应用于流中的每个记录（例如 map()），或者其他运算符将函数一起应用于多个记录（例如 aggregateByKey() ）。您应该查看文档：http://docs.confluent.io/3.0.0/streams/developer-guide.html#kafka-streams-dsl 和示例https://github.com/confluentinc/kafka-streams-examples

因此，您永远不会像上面示例中所示使用 Kafka Streams 创建局部变量，而是将所有内容嵌入链接在一起的运算符/函数中。

例如，如果您想将所有输入记录打印到标准输出，您可以这样做

KStream<String, String> source = builder.stream(stringSerde, stringSerde, "in-stream");
source.foreach(new ForeachAction<String, String>() 
    void apply(String key, String value) 
        System.out.println(key + ": " + value);
    
 );

因此，在您通过streams.start() 启动应用程序后，它将使用来自您输入主题的记录，并且对于您的主题的每条记录，都会调用apply(...)，它将记录打印到标准输出上。

当然，将流打印到控制台的更原生方式是使用source.print()（在内部基本上与显示的foreach() 运算符相同，但已经给定了ForeachAction。）

对于将字符串分配给局部变量的示例，您需要将代码放入 apply(...) 并在那里执行您的正则表达式等以“提取 3 个字母关键字”。

然而，表达这一点的最佳方式是通过flatMapValues() 和print() 的组合（即source.flatMapValues(...).print()）。 flatMapValues() 为每个输入记录调用（在你的情况下，我假设键是 null 所以你可以忽略它）。在您的 flatMapValue 函数中，您应用您的正则表达式，对于每个匹配项，您将匹配项添加到您最终返回的值列表中。

source.flatMapValues(new ValueMapper<String, Iterable<String>>() 
    @Override
    public Iterable<String> apply(String value) 
        ArrayList<String> keywords = new ArrayList<String>();

        // apply regex to value and for each match add it to keywords

        return keywords;
    

flatMapValues 的输出将再次成为KStream，其中包含每个找到的关键字的记录（即，输出流是您在ValueMapper#apply() 中返回的所有列表的“联合”）。最后，您只需通过print() 将结果打印到控制台。 （当然，您也可以使用单个 foreach 而不是 flatMapValue+print，但这会减少模块化。）

【讨论】：

哇。很好的答案伙伴。这就是我要找的东西！

')' 在 foreach 循环结束时丢失。

谢谢@asitm9！更新。下次随意直接编辑:)