Apache Kafka：使用java方式操作stream(实现官方的wordcount)

Posted 2021-09-03 你是小KS

当前版本：kafka_2.12-2.8.0

1. 声明

当前内容主要为使用kafka的stream实现官方的wordcount操作，并将结果输出到控制台，当前内容主要参考：官方文档

2. 基本代码

package com.hy.apache.kafka.start.api.streams;

import java.util.Arrays;
import java.util.Properties;

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.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.Printed;
import org.apache.kafka.streams.kstream.Produced;
import org.apache.kafka.streams.state.KeyValueStore;

/**
 * 
 * @author hy
 * @createTime 2021-05-30 13:54:42
 * @description 当前内容主要为测试和使用当前Kafka的stream的基本测试
 *
 */
public class StreamApiTest {
	public static void main(String[] args) {
		Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-application");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.105: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();
        // 设置stream的来源
        KStream<String, String> textLines = builder.stream("TextLinesTopic");
        KTable<String, Long> wordCounts = textLines
        	// 将读取到的数据行进行处理，按照空格进行分割(即获取到所有的单词)
            .flatMapValues(textLine -> Arrays.asList(textLine.toLowerCase().split("\\\\W+")))
            .groupBy((key, word) -> word)
            // 最后将结果集统计并方在一个counts-store的里面
            .count(Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("counts-store"));
        // 最后将这个数据转换为流并放入一个WordsWithCountsTopic的主题里面
        //wordCounts.toStream().to("WordsWithCountsTopic", Produced.with(Serdes.String(), Serdes.Long()));
        // 将计算后的结果在控制台输出
        wordCounts.toStream().print(Printed.<String, Long>toSysOut());
        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();
      
	}
}

这里将toStream设置print为当前的控制台操作(不是写入到一个topic中)

注意上面的topic必须是已经存在的，否则在执行的时候回报错的

3. 创建topic并向里面写入数据

 ./kafka-console-producer.sh --topic TextLinesTopic --bootstr-server 192.168.1.105:9092

然后开始向里面写入数据：

这里由于警告信息，干扰了写入数据的显示

4. 测试

写入完毕后开启测试结果如下：

只要写入了数据，过一段时间就会显示出来，使用的操作和flink很相似