编写kafka stream应用程序

水深无声 2021-09-21 06:46 327阅读 0赞

关于kafka stream的介绍此处不再多做介绍，可以参考别的博客。直接看代码。

### 第一个stream应用程序Pipe ###

创建一个pipe类：

public class Pipe {
     
        public static void main(String[] args) throws Exception {
     
        }
    }

编写Streams应用程序的第一步是创建一个java.util.Properties映射，以指定StreamsConfig中定义的不同Streams执行配置值。 需要设置几个重要的配置值：StreamsConfig.BOOTSTRAP\_SERVERS\_CONFIG（用于指定用于建立与Kafka集群的初始连接的主机/端口对的列表）和StreamsConfig.APPLICATION\_ID\_CONFIG（用于提供Streams的唯一标识符） 应用程序，以便与与同一个Kafka群集通信的其他应用程序区分开来：

Properties props = new Properties();
    props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-pipe");
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

此外，可以在同一映射中自定义其他配置，例如，记录键值对的默认序列化和反序列化库：

props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
    props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

接下来，我们将定义Streams应用程序的计算逻辑。 在Kafka Streams中，此计算逻辑被定义为连接的处理器节点的拓扑。 我们可以使用拓扑构建器来构建这样的拓扑，

final StreamsBuilder builder = new StreamsBuilder();

然后使用以下拓扑生成器从名为streams-plaintext-input的Kafka topic创建源流：

KStream<String, String> source = builder.stream("streams-plaintext-input");

现在，我们得到了一个KStream，它从其源Kafka主题streams-plaintext-input连续生成记录。 记录被组织为String类型的键值对。 我们可以使用此流执行的最简单的操作是将其写入另一个Kafka主题，即名为streams-pipe-output

source.to("streams-pipe-output");

请注意，我们还可以将上述两行连接为一行，如下所示：

builder.stream("streams-plaintext-input").to("streams-pipe-output");

通过执行以下操作，我们可以检查从此builder创建的拓扑类型：

final Topology topology = builder.build();

并将其描述打印为标准输出为：

System.out.println(topology.describe());

此时，如果我们编译并运行该程序，它将输出以下信息：

Topologies:
       Sub-topology: 0
        Source: KSTREAM-SOURCE-0000000000 (topics: [streams-plaintext-input])
          --> KSTREAM-SINK-0000000001
        Sink: KSTREAM-SINK-0000000001 (topic: streams-pipe-output)
          <-- KSTREAM-SOURCE-0000000000

从上面的打印结果当中，说明了构造的拓扑具有两个处理器节点，即**源节点KSTREAM-SOURCE-0000000000和宿节点KSTREAM-SINK-0000000001**。 KSTREAM-SOURCE-0000000000连续从Kafka主题流-明文输入中读取记录，并将它们通过管道传输到其下游节点KSTREAM-SINK-0000000001; KSTREAM-SINK-0000000001会将接收到的每个记录写入另一个Kafka topic streams-pipe-output（->和<-箭头指示此节点的下游和上游处理器节点，即“子级”和“父级”）。它还说明了这种简单的拓扑没有与之关联的全局状态存储。  
  
请注意，当我们在代码中构建拓扑时，我们始终可以在任何给定的点上像上面一样描述拓扑，因此作为用户，可以交互地“尝试”拓扑中定义的计算逻辑，直到对它满意为止。假设我们已经完成了这种简单的拓扑结构，即以一种无限的流方式将数据从一个Kafka主题传送到另一个主题，现在我们可以使用上面刚刚构建的两个组件来构造Streams客户端：指定的java.util.Properties实例和`Topology` 对象。

final KafkaStreams streams = new KafkaStreams(topology, props);

通过调用其start（）函数，我们可以触发此客户端的执行。 在此客户端上调用close（）之前，执行不会停止。 例如，我们可以添加一个带有倒计时闩锁的关闭hook ，以捕获用户中断并在终止该程序时关闭客户端：

final CountDownLatch latch = new CountDownLatch(1);
     
    // attach shutdown handler to catch control-c
    Runtime.getRuntime().addShutdownHook(new Thread("streams-shutdown-hook") {
        @Override
        public void run() {
            streams.close();
            latch.countDown();
        }
    });
     
    try {
        streams.start();
        latch.await();
    } catch (Throwable e) {
        System.exit(1);
    }
    System.exit(0);

截止到目前我们的代码长这个样子

import org.apache.kafka.common.serialization.Serdes;
    import org.apache.kafka.streams.KafkaStreams;
    import org.apache.kafka.streams.StreamsBuilder;
    import org.apache.kafka.streams.StreamsConfig;
    import org.apache.kafka.streams.Topology;
    import org.apache.kafka.streams.kstream.KStream;
    
    import java.util.Properties;
    import java.util.concurrent.CountDownLatch;
    
    public class Pipe {
        public static void main(String[] args) throws Exception {
            Properties props = new Properties();
            props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-pipe");
            props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
            props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
            props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
    
            final StreamsBuilder builder = new StreamsBuilder();
            KStream<String, String> source = builder.stream("streams-plaintext-input");
            source.to("streams-pipe-output");
    
            final Topology topology = builder.build();
            //System.out.println(topology.describe());
    
            final KafkaStreams streams = new KafkaStreams(topology, props);
            final CountDownLatch latch = new CountDownLatch(1);
    
            // attach shutdown handler to catch control-c
            Runtime.getRuntime().addShutdownHook(new Thread("streams-shutdown-hook") {
                @Override
                public void run() {
                    streams.close();
                    latch.countDown();
                }
            });
    
            try {
                streams.start();
                latch.await();
            } catch (Throwable e) {
                System.exit(1);
            }
            System.exit(0);
        }
    }

如果已经在localhost：9092上启动并运行了Kafka代理，并且在该代理上创建了主题streams-plaintext-input和streams-pipe-output，则可以使用Maven在IDE或命令行中运行此代码。 ：

> mvn clean package
     > mvn exec:java -Dexec.mainClass=myapps.Pipe

### 第二个stream应用程序[Line Split][] ###

我们已经介绍了如何使用两个关键组件构造Streams客户端：StreamsConfig和Topology。 现在，让我们继续通过扩展当前Topology来添加一些实际的处理逻辑。 我们可以先复制现有的Pipe.java类来创建另一个程序:LineSplit

并更改其类名以及应用程序id config以与原始程序区分开：

public class LineSplit {
     
        public static void main(String[] args) throws Exception {
            Properties props = new Properties();
            props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-linesplit");
            // ...
        }
    }

由于每个源流的记录都是字符串类型的键值对，因此我们将值字符串视为文本行，并使用FlatMapValues运算符将其拆分为单词：

KStream<String, String> source = builder.stream("streams-plaintext-input");
    KStream<String, String> words = source.flatMapValues(new ValueMapper<String, Iterable<String>>() {
                @Override
                public Iterable<String> apply(String value) {
                    return Arrays.asList(value.split("\\W+"));
                }
            });

运算符将源流作为输入，并通过按顺序处理其源流中的每个记录并将其值字符串分成单词列表，并将每个单词作为新记录生成到输出，来生成名为单词的新流。 文字流。 这是一个无状态运算符，不需要跟踪任何以前收到的记录或已处理的结果。 请注意，如果您使用的是JDK 8，则可以使用lambda表达式并将上述代码简化为：

KStream<String, String> source = builder.stream("streams-plaintext-input");
    KStream<String, String> words = source.flatMapValues(value -> Arrays.asList(value.split("\\W+")));

最后，我们可以将stream这个词写回到另一个Kafka topic，即streams-linesplit-output。 同样，可以将以下两个步骤串联起来（假设使用了lambda表达式）：

KStream<String, String> source = builder.stream("streams-plaintext-input");
    source.flatMapValues(value -> Arrays.asList(value.split("\\W+")))
          .to("streams-linesplit-output");

如果现在将这种扩展拓扑描述为System.out.println（topology.describe（）），则会得到以下内容：

Topologies:
       Sub-topology: 0
        Source: KSTREAM-SOURCE-0000000000 (topics: [streams-plaintext-input])
          --> KSTREAM-FLATMAPVALUES-0000000002, KSTREAM-FLATMAPVALUES-0000000001
        Processor: KSTREAM-FLATMAPVALUES-0000000002 (stores: [])
          --> KSTREAM-SINK-0000000003
          <-- KSTREAM-SOURCE-0000000000
        Processor: KSTREAM-FLATMAPVALUES-0000000001 (stores: [])
          --> none
          <-- KSTREAM-SOURCE-0000000000
        Sink: KSTREAM-SINK-0000000003 (topic: streams-linesplit-output)
          <-- KSTREAM-FLATMAPVALUES-0000000002

如上所示，新的处理器节点KSTREAM-FLATMAPVALUES-0000000001被注入到原始源节点和宿节点之间的拓扑中。 它以源节点为父节点，宿节点为子节点。 换句话说，由源节点获取的每个记录将首先遍历到要添加的新添加的KSTREAM-FLATMAPVALUES-0000000001节点，结果将生成一个或多个新记录。 它们将继续遍历到接收器节点以写回到Kafka。 请注意，此处理器节点是“无状态”的，因为它没有与任何存储（）相关联。

完整的代码如下所示（假设使用了lambda表达式）：

import org.apache.kafka.common.serialization.Serdes;
    import org.apache.kafka.streams.KafkaStreams;
    import org.apache.kafka.streams.StreamsBuilder;
    import org.apache.kafka.streams.StreamsConfig;
    import org.apache.kafka.streams.Topology;
    import org.apache.kafka.streams.kstream.KStream;
     
    import java.util.Arrays;
    import java.util.Properties;
    import java.util.concurrent.CountDownLatch;
     
    public class LineSplit {
     
        public static void main(String[] args) throws Exception {
            Properties props = new Properties();
            props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-linesplit");
            props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
            props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
            props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
     
            final StreamsBuilder builder = new StreamsBuilder();
     
            KStream<String, String> source = builder.stream("streams-plaintext-input");
            source.flatMapValues(value -> Arrays.asList(value.split("\\W+")))
                  .to("streams-linesplit-output");
     
            final Topology topology = builder.build();
            final KafkaStreams streams = new KafkaStreams(topology, props);
            final CountDownLatch latch = new CountDownLatch(1);
     
            // ... same as Pipe.java above
        }
    }

### 编写wordcount 应用程序 ###

现在，我们进一步采取措施，通过对从源文本流中拆分出的单词进行计数来向topology 添加一些“状态”计算。 按照类似的步骤，让我们基于LineSplit.java类创建另一个程序。

public class WordCount {
     
        public static void main(String[] args) throws Exception {
            Properties props = new Properties();
            props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-wordcount");
            // ...
        }
    }

为了对单词进行计数，我们可以首先修改flatMapValues运算符以将输入的字符全部视为小写（假设使用了lambda表达式）：

source.flatMapValues(new ValueMapper<String, Iterable<String>>() {
                @Override
                public Iterable<String> apply(String value) {
                    return Arrays.asList(value.toLowerCase(Locale.getDefault()).split("\\W+"));
                }
            });

为了进行计数汇总，我们必须首先指定我们要使用groupBy运算符将流键入值字符串。 此运算符生成一个新的分组流，然后可以由count运算符聚合，该操作会在每个分组键上生成一个运行计数：

KTable<String, Long> counts =
    source.flatMapValues(new ValueMapper<String, Iterable<String>>() {
                @Override
                public Iterable<String> apply(String value) {
                    return Arrays.asList(value.toLowerCase(Locale.getDefault()).split("\\W+"));
                }
            })
          .groupBy(new KeyValueMapper<String, String, String>() {
               @Override
               public String apply(String key, String value) {
                   return value;
               }
            })
          // Materialize the result into a KeyValueStore named "counts-store".
          // The Materialized store is always of type <Bytes, byte[]> as this is the format of the inner most store.
          .count(Materialized.<String, Long, KeyValueStore<Bytes, byte[]>> as("counts-store"));

请注意，count运算符具有一个Materialized参数，该参数指定运行计数应存储在名为counts-store的状态存储中。 可以实时查询此Counts存储。

我们还可以将计数KTable的changelog流写回到另一个Kafka topic，即streams-wordcount-output。 由于结果是更改日志流，因此应在启用日志压缩的情况下配置写出到topic ：streams-wordcount-output。 请注意，这次值类型不再是String而是Long，因此默认的序列化类不再适用于将其写入Kafka的情况。 我们需要为Long类型提供重写的序列化方法，否则将抛出运行时异常：

counts.toStream().to("streams-wordcount-output", Produced.with(Serdes.String(), Serdes.Long()));

请注意，为了从主题streams-wordcount-output读取changelog 流，需要将反序列化值设置为org.apache.kafka.common.serialization.LongDeserializer。使用JDK 8中的lambda表达式，则上述代码可以简化为：

KStream<String, String> source = builder.stream("streams-plaintext-input");
    source.flatMapValues(value -> Arrays.asList(value.toLowerCase(Locale.getDefault()).split("\\W+")))
          .groupBy((key, value) -> value)
          .count(Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("counts-store"))
          .toStream()
          .to("streams-wordcount-output", Produced.with(Serdes.String(), Serdes.Long()));

使用System.out.println(topology.describe())来打印打钱topology的描述，可以得到下面这样的输出：

Topologies:
       Sub-topology: 0
        Source: KSTREAM-SOURCE-0000000000 (topics: [TextLinesTopic])
          --> KSTREAM-FLATMAPVALUES-0000000001
        Processor: KSTREAM-FLATMAPVALUES-0000000001 (stores: [])
          --> KSTREAM-KEY-SELECT-0000000002
          <-- KSTREAM-SOURCE-0000000000
        Processor: KSTREAM-KEY-SELECT-0000000002 (stores: [])
          --> KSTREAM-FILTER-0000000005
          <-- KSTREAM-FLATMAPVALUES-0000000001
        Processor: KSTREAM-FILTER-0000000005 (stores: [])
          --> KSTREAM-SINK-0000000004
          <-- KSTREAM-KEY-SELECT-0000000002
        Sink: KSTREAM-SINK-0000000004 (topic: counts-store-repartition)
          <-- KSTREAM-FILTER-0000000005
    
      Sub-topology: 1
        Source: KSTREAM-SOURCE-0000000006 (topics: [counts-store-repartition])
          --> KSTREAM-AGGREGATE-0000000003
        Processor: KSTREAM-AGGREGATE-0000000003 (stores: [counts-store])
          --> KTABLE-TOSTREAM-0000000007
          <-- KSTREAM-SOURCE-0000000006
        Processor: KTABLE-TOSTREAM-0000000007 (stores: [])
          --> KSTREAM-SINK-0000000008
          <-- KSTREAM-AGGREGATE-0000000003
        Sink: KSTREAM-SINK-0000000008 (topic: WordsWithCountsTopic)
          <-- KTABLE-TOSTREAM-0000000007

正如我们在上面看到的那样，topology 现在包含两个互相无连接的的sub-topologies。第一个sub-topology的宿节点KSTREAM-SINK-0000000004将写入分区topic Counts-repartition，第二个sub-topology的源节点KSTREAM-SOURCE-0000000006将读取该分区。此外，在第一个sub-topology内部，将无状态的KSTREAM-FILTER-0000000005节点注入到分组的KSTREAM-KEY-SELECT-0000000002节点与接收器节点之间，以过滤掉聚合键为空的任何中间记录。

在第二个子拓扑中，聚合节点KSTREAM-AGGREGATE-0000000003与名为Counts的状态存储关联（名称由用户在count运算符中指定）。从上游节点接收到每个记录后，聚合处理器将首先查询其关联的Counts存储，以获取该键的当前计数，再增加一个，然后将新计数写回到存储中。密钥的每个更新计数也将通过管道传递到KTABLE-TOSTREAM-0000000007节点，该节点将此更新流解释为记录流，然后再进一步传递给接收器节点KSTREAM-SINK-0000000008以写回Kafka。

完整的代码如下所示（假设使用了lambda表达式）：

import org.apache.kafka.common.serialization.Serdes;
    import org.apache.kafka.common.utils.Bytes;
    import org.apache.kafka.streams.KafkaStreams;
    import org.apache.kafka.streams.StreamsBuilder;
    import org.apache.kafka.streams.StreamsConfig;
    import org.apache.kafka.streams.Topology;
    import org.apache.kafka.streams.kstream.KStream;
    import org.apache.kafka.streams.kstream.KTable;
    import org.apache.kafka.streams.kstream.Materialized;
    import org.apache.kafka.streams.kstream.Produced;
    import org.apache.kafka.streams.state.KeyValueStore;
    
    import java.util.Arrays;
    import java.util.Properties;
    import java.util.concurrent.CountDownLatch;
    
    public class WordCount {
    
        public static void main(final String[] args) throws Exception {
            Properties props = new Properties();
            props.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-application");
            props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-broker1:9092");
            props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
            props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
    
            StreamsBuilder builder = new StreamsBuilder();
            KStream<String, String> textLines = builder.stream("TextLinesTopic");
            KTable<String, Long> wordCounts = textLines
                    .flatMapValues(textLine -> Arrays.asList(textLine.toLowerCase().split("\\W+")))
                    .groupBy((key, word) -> word)
                    .count(Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("counts-store"));
            wordCounts.toStream().to("WordsWithCountsTopic", Produced.with(Serdes.String(), Serdes.Long()));
    
            final Topology topology = builder.build();
    
            System.out.println(topology.describe());
    
            final KafkaStreams streams = new KafkaStreams(topology, props);
            final CountDownLatch latch = new CountDownLatch(1);
    
            // attach shutdown handler to catch control-c
            Runtime.getRuntime().addShutdownHook(new Thread("streams-shutdown-hook") {
                @Override
                public void run() {
                    streams.close();
                    latch.countDown();
                }
            });
    
            try {
                streams.start();
                latch.await();
            } catch (Throwable e) {
                System.exit(1);
            }
            System.exit(0);
        }
    
    }

[Line Split]: http://kafka.apache.org/25/documentation/streams/tutorial#tutorial_code_linesplit