什么是 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<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()));
                  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 的一些主要应用程序。