内容简介:欢迎关注公众号:n平方本文主要介绍【Kafka Streams编写关键任务实时应用程序和微服务的最简单方法,是一个用于构建应用程序和微服务的客户端库,其中输入和输出数据存储在Kafka集群中。它结合了在客户端编写和部署标准Java和Scala应用程序的简单性和Kafka服务器端集群技术的优点。
欢迎关注公众号:n平方
本文主要介绍【 KafkaStreams 】
简介
Kafka Streams编写关键任务实时应用程序和微服务的最简单方法,是一个用于构建应用程序和微服务的客户端库,其中输入和输出数据存储在Kafka集群中。它结合了在客户端编写和部署标准 Java 和Scala应用程序的简单性和Kafka服务器端集群技术的优点。
Kafka Streams是一个用于构建关键任务实时应用程序和微服务的客户端库,其中输入和/或输出数据存储在Kafka集群中。Kafka Streams结合了在客户端编写和部署标准Java和Scala应用程序的简单性和Kafka服务器端集群技术的优点,使这些应用程序具有高度可伸缩性、灵活性、容错性、分布式等等。
目标
- 了解kafka Streams
- 会使用kafka Streams
过程
1.首先WordCountDemo示例代码(Java8以上)
// Serializers/deserializers (serde) for String and Long types final Serde<String> stringSerde = Serdes.String(); final Serde<Long> longSerde = Serdes.Long(); // Construct a `KStream` from the input topic "streams-plaintext-input", where message values // represent lines of text (for the sake of this example, we ignore whatever may be stored // in the message keys). KStream<String, String> textLines = builder.stream("streams-plaintext-input", Consumed.with(stringSerde, stringSerde); KTable<String, Long> wordCounts = textLines // Split each text line, by whitespace, into words. .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+"))) // Group the text words as message keys .groupBy((key, value) -> value) // Count the occurrences of each word (message key). .count() // Store the running counts as a changelog stream to the output topic. wordCounts.toStream().to("streams-wordcount-output", Produced.with(Serdes.String(), Serdes.Long()));
它实现了WordCount算法,该算法从输入文本计算单词出现的直方图。然而,与您以前可能看到的对有界数据进行操作的其他WordCount示例不同,WordCount演示应用程序的行为略有不同,因为它被设计为对无限、无界的数据流进行操作。与有界变量类似,它是一种有状态算法,用于跟踪和更新单词的计数。然而,由于它必须假定输入数据可能是无界的,因此它将周期性地输出当前状态和结果,同时继续处理更多的数据,因为它不知道何时处理了“所有”输入数据。
2.安装并启动zookeeper和kafka
bin/zookeeper-server-start.sh config/zookeeper.properties bin/kafka-server-start.sh config/server.properties
3.创建主题
接下来,我们创建名为streams-plain -input的输入主题和名为streams-wordcount-output的输出主题:
bin/kafka-topics.sh --create \ --zookeeper localhost:2181 \ --replication-factor 1 \ --partitions 1 \ --topic streams-plaintext-input Created topic "streams-plaintext-input"
我们创建启用压缩的输出主题,因为输出流是一个变更日志流.
bin/kafka-topics.sh --create \ --zookeeper localhost:2181 \ --replication-factor 1 \ --partitions 1 \ --topic streams-wordcount-output \ --config cleanup.policy=compact Created topic "streams-wordcount-output"
创建的主题也可以使用相同的kafka主题进行描述
bin/kafka-topics.sh --zookeeper localhost:2181 --describe
4.启动Wordcount应用程序
bin/kafka-run-class.sh org.apache.kafka.streams.examples.wordcount.WordCountDemo
a)演示应用程序将从输入主题流(明文输入)中读取,对每个读取的消息执行WordCount算法的计算,并不断将其当前结果写入输出主题流(WordCount -output)。因此,除了日志条目之外,不会有任何STDOUT输出,因为结果是用Kafka写回去的。
b)现在我们可以在一个单独的终端上启动控制台生成器,向这个主题写入一些输入数据和检查输出的WordCount演示应用程序从其输出主题与控制台消费者在一个单独的终端.
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \ --topic streams-wordcount-output \ --from-beginning \ --formatter kafka.tools.DefaultMessageFormatter \ --property print.key=true \ --property print.value=true \ --property key.deserializer=org.apache.kafka.common.serialization.StringDeserializer \ --property value.deserializer=org.apache.kafka.common.serialization.LongDeserializer
c) 输入端: 现在让我们使用控制台生成器将一些消息写入输入主题流——纯文本输入,方法是输入一行文本,然后单击。这将发送新消息输入主题,消息键为空和消息值是刚才输入的字符串编码的文本行。
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic streams-plaintext-input
此时你可以在 控制台 输入如下字符:
all streams lead to kafka
d)) 输出端: 此消息将由Wordcount应用程序处理,以下输出数据将写入streams-wordcount-output主题并由控制台使用者打印:
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \ --topic streams-wordcount-output \ --from-beginning \ --formatter kafka.tools.DefaultMessageFormatter \ --property print.key=true \ --property print.value=true \ --property key.deserializer=org.apache.kafka.common.serialization.StringDeserializer \ --property value.deserializer=org.apache.kafka.common.serialization.LongDeserializer
这个时候会接收到刚刚在 控制台 输入的单词统计结果:
all 1 streams 1 lead 1 to 1 kafka 1
如此类推:你可以在 输入端 输入单词,对应的在 输出端 就会有统计结果。
小结:
可以看到,Wordcount应用程序的输出实际上是连续的更新流,其中每个输出记录(即上面原始输出中的每一行)是单个单词的更新计数,也就是记录键,如“kafka”。对于具有相同键的多个记录,后面的每个记录都是前一个记录的更新。
下面的两个图说明了幕后的本质。第一列显示KTable的当前状态的演变,该状态为count计算单词出现的次数。第二列显示KTable的状态更新所产生的更改记录,这些记录被发送到输出Kafka主题流-wordcount-output。
最后
本人水平有限,欢迎各位建议以及指正。顺便关注一下公众号呗,会经常更新文章的哦。
以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持 码农网
猜你喜欢:本站部分资源来源于网络,本站转载出于传递更多信息之目的,版权归原作者或者来源机构所有,如转载稿涉及版权问题,请联系我们。
现代操作系统(第3版)
Andrew S. Tanenbaum / 陈向群、马洪兵 / 机械工业出版社 / 2009-7 / 75.00元
本书是操作系统领域的经典之作,与第2版相比,增加了关于Linux、Windows Vista和Symbian操作系统的详细介绍。书中集中讨论了操作系统的基本原理,包括进程、线程、存储管理、文件系统、输入/输出、死锁等,同时还包含了有关计算机安全、多媒体操作系统、掌上计算机操作系统、微内核、多核处理机上的虚拟机以及操作系统设计等方面的内容。此外,还在第2版的基础上对部分习题进行了增删,更有助于读者学......一起来看看 《现代操作系统(第3版)》 这本书的介绍吧!