什么是 Kafka Streams 以及它们是如何实现的？

Apache Kafka Streams API 是一个开源、健壮、一流、水平可扩展的消息传递系统。通俗地说，它是一个升级后的 Kafka 消息系统，构建在Apache Kafka之上。在本文中，我们将通过以下记录了解它到底是什么。

What is Kafka?

What is a Stream?

What exactly is Kafka Stream?

Apache Kafka Stream API Architecture

Kafka Stream Features

Kafka Streams Example

Differences between Kafka and Kafka Streams

Use cases of Apache Kafka Streams API

1、What is Kafka?

Apache Kafka基本上是由 Linkedin 开发的开源消息传递工具，为实时数据馈送提供低延迟和高吞吐量的平台。它是使用Scala 和Java编程语言开发的。

2、什么是流？

通常，Stream 可以定义为无界且连续的实时数据包流。数据包以键值对的形式生成，这些是从发布者自动转移的，无需提出相同的请求。

3、Kafka Stream 究竟是什么？

Apache Kafka Stream 可以定义为一个开源客户端库，用于构建应用程序和微服务。在这里，输入和输出数据存储在 Kafka Clusters 中。它将设计和部署标准 Scala 和Java 应用程序的可理解性与 Kafka 服务器端集群的优势相结合技术。

Apache Kafka Stream API 架构

Apache KStreams 内部使用生产者和消费者库。它基本上与 Kafka 相结合，API 允许您通过实现数据并行、容错和许多其他强大功能来利用 Kafka 的能力。

KStream 架构中存在的不同组件如下：

输入流

输出流

实例

消费者

本地状态

流拓扑

Input Stream 和 Output Streams 是 Kafka 集群，用于存储所提供任务的输入和输出数据。

在每个实例中，我们都有消费者、流拓扑和本地状态

流拓扑实际上是执行给定任务的流或 DAG

本地状态是存储给定操作（如 Map、FlatMap 等）的中间结果的内存位置。

为了增加数据并行度，我们可以直接增加Instances的数量。继续前进，我们将了解 Kafka Streams 的功能。

Kafka 流特性

现在，让我们讨论 Kafka 流的重要特性，这些特性使其比其他类似技术更具优势。

松紧带

Apache Kafka 是一个开源项目，旨在提供高可用性和水平可扩展性。因此，在 Kafka 的支持下，Kafka Streams API 实现了它的高弹性和易于扩展的特性。

容错

数据日志最初是分区的，这些分区在集群中处理数据和相应请求的所有服务器之间共享。因此，Kafka 通过在多个服务器上复制每个分区来实现容错。

高度可行

由于 Kafka 集群具有高可用性，因此无论其大小如何，它们都可以成为任何类型的用例的首选。它们能够支持小型、中型和大型用例。

综合安全

Kafka 具有三个主要安全组件，可为集群中的数据提供一流的安全性。它们如下所述：

使用 SSL/TLS 加密数据

SSL/SASL 认证

ACL 的授权

其次是安全性，我们有它对高端编程语言的支持。

支持 Java 和 Scala

Kafka Streams API 最好的部分是它自身集成了 Java 和 Scala 等最主要的编程语言，并且可以轻松设计和部署 Kafka 服务器端应用程序。

恰好一次处理语义

通常，流处理是对无限系列的数据或事件的连续执行。但在卡夫卡的情况下，事实并非如此。Exactly-Once 意味着用户定义的语句或逻辑只执行一次，对状态的更新，由 SPE（流处理元素）管理，在一个持久的时间里只提交一次e后台商店

Kafka 流示例

可以使用 Java 编程语言执行此特定示例。然而，这有一些先决条件。需要在本地系统中安装Kafka和Zookeeper。

代码是为wordcount编写的，其记录如下。

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.Produced; import org.apache.kafka.streams.state.KeyValueStore; import java.util.Arrays; import java.util.Properties; public class WordCountApplication { 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 textLines = builder.stream("TextLinesTopic"); KTable wordCounts = textLines.flatMapValues(textLine -> Arrays.asList(textLine.toLowerCase().split("W+"))).groupBy((key, word) -> word).count(Materialized.>as("counts-store")); wordCounts.toStream().to("WordsWithCountsTopic", Produced.with(Serdes.String(), Serdes.Long())); KafkaStreams streams = new KafkaStreams(builder.build(), props); streams.start(); } }

现在，我们将了解 Kafka 和 Kafka Streams 之间的一些重要区别。

//给出的文本

欢迎来到 Edureka Kafka 培训。

这篇文章是关于 Kafka Streams 的。

//输出：

Welcome(1)

to(1)

Edureka(1)

Kafka(2)

Training(1)

This(1)

article(1)

is(1)

about(1)

Streams(1)

Kafka 和 Kafka Streams 之间的差异

Apache Kafka Streams API 用例

Apache Streams API 用于多个用例。下面提到了使用 Streams API 的一些主要应用程序。

Apache Kafka

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。