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

Question

我一直在为我正在处理的 java 应用程序查找大量 Kafka 文档。我已经尝试了解 Java 8 中引入的 lambda 语法，但我在这方面有点粗略，并且不太确定它应该是我目前使用的语法。

我有一个 Kafka/Zookeeper 服务运行没有任何问题，我想做的是编写一个小示例程序，根据输入将其写出来，但不做wordcount 因为已经有很多例子了。

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

示例数据

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、生产者和消费者运行都连接到同一个主题，所以我想基本上看到相同的 String 出现在所有实例（生产者、消费者和流）。

Answer 1

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

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

因此，您永远不会像上面示例中所示那样使用 Kafka Streams 创建局部变量，而是将所有内容嵌入 operators/functions 中并链接在一起。

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

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(...) 并在 "extract the 3 letter keywords".[=36= 中执行正则表达式等操作]

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

的值列表中

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，包含每个找到的关键字的记录（即，输出流是 "union" 所有列表 return 在 ValueMapper#apply() 中）。最后，您只需通过 print() 将结果打印到控制台。（当然，您也可以使用单个 foreach 而不是 flatMapValue+print，但这会降低模块化程度。）

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

Print Kafka Stream Input out to console?

java

stream

apache-kafka

apache-kafka-streams

示例数据

问题

代码