WordCount
WordCount 堪称大数据界的HelloWorld,相信不管是Hadoop还是Spark等大数据工具的上手实例,第一个十有八九是WordCount。
Kafka Stream也不例外。作为集成在Kafka消息系统上的数据实时处理接口,WordCount也可以作为一个很好的入门实例。
实际上,Kafka官方已经提供了WordCount的Demo,org.apache.kafka.streams.examples.wordcount.WordCountDemo
,但亲手实现一遍可以帮助我们快速入门。
逻辑流程
需要记住的是,Kafka中的数据都以<key, value>的形式存在。
假设我们的Kafka中,已经存在一个topic,其中的数据来自于一个文本文件。我们希望编写一个Kafka Streams Application对此topic中的数据进行WordCount计算,大概步骤可以分解为:
Stream 从源topic中取出每一行数据记录 (<key, value>格式) —- <null, “Hello World hello”>
MapValue 将value中所有文本转换成小写形式 —- <null, “hello world hello”>
FlatMapValues 按空格分解成单词 —- <null, “hello”>,<null, “world”>, <null, “hello”>
SelectKey 将value的值赋给key —- <“hello”, “hello”>,<“world”, “world”>, <“hello”, “hello”>
GroupByKey 按相同的Key分组 —- (<“hello”, “hello”>, <“hello, “hello”>),(<“world”, “world”>)
Count 计算每个组中元素个数 —- <“hello”, 2>,<“world”, 1>
To 将结果返回Kafka
逻辑代码
首先进行配置,包括Kafka Streams Application的ID,Kafka集群位置等:
Properties config = new Properties();
config.put(StreamsConfig.APPLICATION_ID_CONFIG, "word-count");
config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
这里使用Kafka Streams DSL。DSL提供的各种算子大部分情况下都可以满足需求。
使用DSL:
StreamsBuilder builder = new StreamsBuilder();
从Kafka源topic获取数据流:
KStream<String, String> textLines = builder.stream("streams-plaintext-input");
KStream即代表了由各个数据记录组成的数据流。
KStream可以从一或更多topic中的数据得来。
KStream可以进行对数据记录的逐条转换,和其它KStream,KTable进行join操作,或aggregate成KTable。
对得到的KStream进行transformation和aggregation:
将数据记录中的大写全部替换成小写:
.mapValues(textLine -> textLine.toLowerCase())
将各行数据按空格拆分:
.flatMapValues(textLine -> Arrays.asList(textLine.split(" ")))
将value作为新的key:
.selectKey((key, word) -> word)
aggregation操作前group by key:
.groupByKey()
计算每个组中的元素个数:
.count(Materialized.as("Counts"));
得到结果后将其存储为KTable:
KTable<String, Long> wordCounts = textLines
.mapValues ...
最后导入目标topic,其中key为String,value为Long。
wordCounts.toStream().to("streams-wordcount-output", Produced.with(Serdes.String(), Serdes.Long()));
KafkaStreams streams = new KafkaStreams(builder.build(), config);
streams.start();