14

我一直在查看我正在开发的 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所有实例（生产者、消费者和流）上都出现相同的内容。

4

1 回答 1

28

如果您使用 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-流示例

因此，您永远不会像上面示例中所示使用 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()为每个输入记录调用（在你的情况下，我假设 key 是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再次为 a，包含每个找到的关键字的记录（即，输出流是您返回的所有列表的“联合” ValueMapper#apply()）。最后，您只需通过 将结果打印到控制台print()。（当然，您也可以使用单个foreach而不是flatMapValue+ print，但这会减少模块化。）

于 2016-09-05T22:59:21.773 回答