将 Kafka 流输入打印到控制台？示例数据问题法典

java stream apache-kafka apache-kafka-streams

2022-09-03 18:23: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 = ?
}

我有动物园管理员，kafka，生产者和消费者运行都连接到同一个主题，所以我希望基本上看到相同的出现在所有实例（生产者，消费者和流）上。String

答案 1

如果使用 Kafka Streams，则需要在数据流上应用函数/运算符。在本例中，您创建了一个对象，因此，您希望将运算符应用于。KStreamsource

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

因此，您永远不会使用 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);
    }
 });

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

当然，将流打印到控制台的更原生的方法是使用（这在内部基本上与所示的运算符相同，并已给出一个。source.print()foreach()ForeachAction

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

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

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;
    }
}

的输出将再次是一个，包含每个找到的关键字的记录（即，输出流是所有列表的“联合” 返回）。最后，您只需通过将结果打印到控制台即可。（当然，您也可以使用单个而不是+，但这将不那么模块化。flatMapValuesKStreamValueMapper#apply()print()foreachflatMapValueprint

答案 2