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

Question

我一直在查阅 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.

问题

我想要提取三个字母的关键词并用 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。

- Zeliax

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(...)，并在 stdout 上打印该记录。

当然，更本地的将流打印到控制台的方法是使用 source.print()（内部基本上与已给出的带有 ForeachAction 的 foreach() 操作符相同）。

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

然而，最好的表达方式是通过 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代替flatMapValues+print，但这样会更少模块化。）

- Matthias J. Sax

---

哇，伙计你真给力。这就是我一直在找的！ - Zeliax

foreach循环缺少')'。 - asitm9

谢谢@asitm9！已更新。下次请随意直接编辑 :) - Matthias J. Sax