Kafka：Streams

关于

Kafka Streams 是用于构建应用程序和微服务的客户端库，其中输入和输出数据存储在Kafka集群中。 它结合了在客户端编写和部署标准Java和Scala应用程序的简便性以及Kafka服务器端集群技术的优势。

Kafka Streams是一个客户端程序库，用于处理和分析存储在Kafka中的数据，并将得到的数据写回Kafka或发送到外部系统。

Kafka Stream基于一个重要的流处理概念。如正确的区分事件时间和处理时间，窗口支持，以及简单而有效的应用程序状态管理。Kafka Streams的入口门槛很低: 你可以快速的编写和在单台机器上运行一个小规模的概念证明（proof-of-concept）；而你只需要运行你的应用程序部署到多台机器上，以扩展高容量的生产负载。

Kafka Stream利用kafka的并行模型来透明的处理相同的应用程序作负载平衡。

Kafka Stream 的特点：

• 设计一个简单的、轻量级的客户端库，可以很容易地嵌入在任何java应用程序与任何现有应用程序封装集成。
• Apache Kafka本身作为内部消息层，没有外部系统的依赖，还有，它使用kafka的分区模型水平扩展处理，并同时保证有序。
• 支持本地状态容错，非常快速、高效的状态操作（如join和窗口的聚合）。
• 采用 one-recored-at-a-time（一次一个消息） 处理以实现低延迟，并支持基于事件时间(event-time)的窗口操作。
• 提供必要的流处理原语(primitive)，以及一个 高级别的Steram DSL 和 低级别的Processor API。

核心概念

Stream 处理拓扑

Stream 处理拓扑：

• 流 是Kafka Stream提出的最重要的抽象概念：它表示一个无限的，不断更新的数据集。
  • 流是一个有序的，可重放（反复的使用），不可变的容错序列，数据记录的格式是键值对（key-value）。
• 通过 Kafka Streams 编写一个或多个的计算逻辑的处理器拓扑。
  • 其中处理器拓扑是一个由流（边缘）连接的流处理（节点）的图。
• 流处理器是处理器拓扑中的一个节点；它表示一个处理的步骤，用来转换流中的数据（从拓扑中的上游处理器一次接受一个输入消息，并且随后产生一个或多个输出消息到其下游处理器中）。

在拓扑中有两个特别的处理器：

1. 源处理器（Source Processor）：源处理器是一个没有任何上游处理器的特殊类型的流处理器。
   • 它从一个或多个kafka主题生成输入流。通过消费这些主题的消息并将它们转发到下游处理器。
2. Sink处理器：sink处理器是一个没有下游流处理器的特殊类型的流处理器。
   • 它接收上游流处理器的消息发送到一个指定的Kafka主题。

Kafka streams 提供 2 种方式来定义流处理器拓扑：

1. Kafka Streams DSL 提供了更常用的数据转换操作，如 map 和 filter；
2. 低级别Processor API允许开发者定义和连接自定义的处理器，以及和状态仓库交互。

处理器拓扑仅仅是流处理代码的逻辑抽象。

时间

时间 在流中的常见概念如下：

1. 事件时间：当一个事件或数据记录发生的时间点，就是最初创建的“源头”。
2. 处理时间：事件或数据消息发生在流处理应用程序处理的时间点。即，记录已被消费。
   • 处理时间可能是毫秒，小时，或天等。比原始事件时间要晚。
3. 摄取时间：事件或数据记录是 Kafka broker 存储在topic分区的时间点。
   • 与事件时间的差异是，当记录由 Kafka broker 追加到目标topic时，生成的摄取时间戳，而不是消息创建时间（“源头”）。
   • 与处理时间的差异是处理时间是流处理应用处理记录时的时间。比如，如果一个记录从未被处理，那么久没有处理时间，但仍然有摄取时间。

Kafka Streams 通过 TimestampExtractor 接口为每个数据记录分配一个时间戳：

该接口的具体实现了基于数据记录的实际内容检索或计算获得时间戳，例如嵌入时间戳字段提供的事件时间语义，或使用其他的方法，比如在处理时返回当前的 wall-clock（墙钟）时间，从而产生了流应用程序的处理时间语义。

因此开发者可以根据自己的业务需要选择执行不同的时间。

例如，每条记录时间戳描述了流的时间增长（尽管记录在stream中是无序的）并利用时间依赖性来操作，如 join。

最后，当一个 Kafka Streams 应用程序写入记录到 kafka 时，它将分配时间戳到新的消息。时间戳分配的方式取决于上下文：

• 当通过处理一些输入记录（例如，在process()函数调用中触发的context.forward()）生成新的输出记录时，输出记录时间戳直接从输入记录时间戳继承。
• 当通过周期性函数（如punctuate()）生成新的输出记录时，输出记录时间戳被定义为流任务的当前内部时间（通过context.timestamp()获取）。
• 对于聚合，生成的聚合更新的记录时间戳将被最新到达的输入记录触发更新。

状态

一些流处理程序不需要状态，这意味着消息处理是独立于其他的消息处理的。但是，能够保持状态，这为复杂的流处理程序打开了许多可能性：你可以加入输入流，或分组和汇总数据记录等。

• Streams DSL提供了许多如状态性的操作。

Kafka Stream 提供了所谓的状态存储，流处理程序可以用来存储和查询数据：

在Kafka Stream中的每一个任务嵌入了一个或多个状态存储，可通过API来存储和查询处理所需的数据。

状态存储可以是一个持久的 key/value 存储，内存中的 HashMap，或者是其他的数据结构。

• Kafka Stream 提供了本地状态存储的故障容错和自动恢复。

ARCHITECTURE（架构）

Kafka Streams通过生产者和消费者，并利用kafka自有的能力来提供数据平行性，分布式协调性，故障容错和操作简单性，从而简化了应用程序的开发。

Stream分区和任务

Kafka分区数据的消息层用于存储和传输。Kafka Streams分区数据用于处理。 在这两种情况下，这种分区使数据弹性，可扩展，高性能和容错。

Kafka Streams使用了分区和任务的概念，基于Kafka主题分区的并行性模型。

在并发环境行，Kafka Streams和Kafka之间有着紧密的联系：

• 每个流分区是完全有序的数据记录队列，并映射到kafka主题的分区。
• 流的数据消息与主题的消息映射。
• 数据记录中的 keys 决定了 Kafka 和 Kafka Streams 中数据的分区，即，如何将数据路由到指定的分区。

应用程序的处理器拓扑通过将其分成多个任务来进行扩展，更具体点说，Kafka Streams 根据输入流分区创建固定数量的任务，其中每个任务分配一个输入流的分区列表（即，Kafka主题）。分区对任务的分配不会改变，因此每个任务是应用程序并行性的固定单位。然后，任务可以基于分配的分区实现自己的处理器拓扑；他们还可以为每个分配的分区维护一个缓冲，并从这些记录缓冲一次一个地处理消息。作为结果，流任务可以独立和并行的处理而无需手动干预。

重要的是要理解 Kafka Streams 不是资源管理器，而是可在任何地方都能“运行”的流处理应用程序库。多个实例的应用程序在同一台机器上执行，或分布多个机器上，并且任务可以通过该库自动的分发到这些运行的实例上。

• 分区对任务的分配永远不会改变；如果一个应用程式实例失败，则其被分配的任务将自动地在其他的实例重新创建，并从相同的流分区继续消费。

下面展示了2个分区，每个任务分配了输出流的1个分区：

线程模型

Kafka Streams允许用户配置线程数，可用于平衡处理应用程序的实例。每个线程的处理器拓扑独立的执行一个或多个任务。

例如，下面展示了一个流线程运行2个流任务：

启动更多的流线程或更多应用程序实例，只需复制拓扑逻辑（ps，就是多复制几个代码到不同的机器上运行），达到并行处理处理不同的Kafka分区子集的目的。

• 要注意的是，这些线程之间不共享状态。因此无需协调内部的线程。这使它非常简单在应用实例和线程之间并行拓扑。Kafka主题分区的分配是通过Kafka Streams利用Kafka的协调功能在多个流线程之间透明处理。

如上所述，Kafka Streams扩展流处理应用程序是很容易的：你只需要运行你的应用程序实例，Kafka Streams负责在实例中运行的任务之间分配分区。你可以启动和应用程序线程一个多的输入Kafka主题分区。这样，所有运行中的应用实例，每个线程（或更确切的说，它运行的任务）至少有一个输入分区可以处理。

本地状态存储

存储，其实是流处理器应用程序可用来存储和查询数据，对于实现状态性操作是一个很重要的能力。

例如，当你调用状态性操作时，如 join() 或 aggregate()，或当你在窗口化流时，Kafka Streams DSL 会自动创建和管理这些状态存储。

在Kafka Streams应用程序的每个流任务可以键入一个或多个本地状态存储，这些本地状态存储可以通过API存储和查询处理所需的数据。Kafka Streams也为本地状态存储提供了容错和自动恢复的能力。

下图显示了两个流任务及其专用本地状态存储：

故障容错

Kafka Streams 基于 Kafka 分区的高可用和副本故障容错能力。因此，当流数据持久到Kafka，即使应用程序故障，如果需要重新处理它，它也是可用的。Kafka Streams中的任务利用Kafka消费者客户端提供的故障容错的能力来处理故障。如果任务故障，Kafka Streams将自动的在剩余运行中的应用实例重新启动该任务。

此外，Kafka Streams还确保了本地状态仓库对故障的稳定性。对于每个状态仓库都维持一个追踪所有的状态更新的变更日志主题。这些变更日志主题也分区，因此，每个本地状态存储实例，任务访问仓里，都有自己的专用的变更日志分区。变更主题日志也启用了日志压缩，以便可以安全的清除旧数据，以防止主题无限制的增长。如果任务失败并在其他的机器上重新运行，则Kafka Streams在恢复新启动的任务进行处理之前，重放相应的变更日志主题，保障在故障之前将其关联的状态存储恢复。故障处理对于终端用户是完全透明的。

请注意，任务（重新）初始化的成本通常主要取决于通过重放状态仓库变更日志主题来恢复状态的时间。为了减少恢复时间，用户可以配置他们的应用程序增加本地状态的备用副本（即。完全的复制状态）。当一个任务迁移发生时，Kafka Streams 尝试去分配任务给应用实例。