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

Question

我一直在查看我正在处理的 Java 应用程序的大量 Kafka 文档。我已经尝试了解 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、生产者和消费者都连接到同一个主题，所以我想基本上看到相同的 String 出现在所有实例(生产者、消费者和流)上。

Answer 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(.. .).打印()). 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 但这会降低模块化。)