Kafka 实战 - Kafka Streams 基本概念及API详解

Apache Kafka Streams 是一个用于构建实时流处理应用程序的客户端库，它直接构建在 Kafka 之上，允许开发者使用纯 Java 或 Scala 代码轻松处理 Kafka 中的数据流。以下是对 Kafka Streams 的基本概念及 API 的详解：

基本概念

**1. **流（Stream）

流是 Kafka Streams 最基础的抽象，它代表一个无限的、持续更新的数据集。每个流由一系列带有键值对的不可变记录组成，这些记录按照其在流中的位置（即偏移量）排序。流中的数据可以被重播，因为它持久存储在 Kafka 主题中，并且支持故障转移，使得流处理应用程序能够可靠地处理数据。

**2. **KStream 和 KTable

Kafka Streams 提供了两种核心的流处理抽象：

• KStream: 用于表示无界、持续更新的数据流，类似于传统的流处理概念。KStream 可以执行过滤、映射、聚合等操作，也可以与其他 KStream 或 KTable 进行连接（join）。
• KTable: 代表一个随时间变化的、键值对形式的表。KTable 适合存储具有唯一键的、随着时间推移可能会发生更新的数据。当新记录到达时，KTable 会更新其内部状态，反映最新的键值对。KTable 支持连接（join）和聚合操作，特别适用于实现基于事件时间的窗口化聚合。

**3. **时间概念

Kafka Streams 支持两种时间概念：

• 处理时间（Processing Time）: 当前系统时间，即流处理应用程序在处理记录时的实际时间。
• 事件时间（Event Time）: 记录中携带的时间戳，代表事件在其源头发生的时间。事件时间允许处理乱序事件和实现精确的基于时间窗口的聚合。

**4. **窗口（Window）

窗口是流处理中用来对无限数据流进行有限处理的重要概念。Kafka Streams 提供了时间窗口（如固定窗口、滑动窗口、会话窗口）和计数窗口，允许在特定时间段内或特定数量的记录范围内对数据进行聚合。

**5. **状态管理

Kafka Streams 具有内置的状态管理能力，允许应用程序维护有状态的处理逻辑。状态可以是局部的（每个实例有自己的状态副本）或全局的（所有实例共享状态）。状态存储在本地（通常使用 RocksDB），并通过 Kafka 的复制机制实现容错。

Kafka Streams API详解

Kafka Streams API 主要分为两个层次：

**1. **高级 DSL（Domain Specific Language）

提供了易于使用的高层接口，通过一系列操作符（如

map

,

filter

,

groupByKey

,

reduce

,

join

,

windowedBy

等）构建流处理逻辑。DSL 使开发者无需关心底层细节，可以快速编写简洁、声明式的流处理代码。

例如：

KStream<String,String> textLines = builder.stream("input-topic");KStream<String,Long> wordCounts = textLines
    .flatMapValues(line ->Arrays.asList(line.toLowerCase().split("\\W+"))).filter((key, word)->!word.isEmpty()).groupBy((key, word)-> word).count(Materialized.as("counts-store"));
wordCounts.to("output-topic",Produced.with(Serdes.String(),Serdes.Long()));

这段代码定义了一个从

input-topic

读取文本行，进行单词拆分、过滤、按单词分组并计数，最后将结果写入

output-topic

的流处理任务。

**2. **低级 Processor API

对于更复杂或定制化的场景，可以使用 Processor API 直接定义处理器（

Processor

、

Transformer

、

Punctuator

等）和处理器链。Processor API 提供了对流处理过程的细粒度控制，但要求开发者自行管理状态存储、定时器等。

例如，定义一个自定义

Transformer

：

classMyTransformerimplementsTransformer<String,String,KeyValue<String,String>>{ProcessorContext context;KeyValueStore<String,Integer> store;@Overridepublicvoidinit(ProcessorContext context){this.context = context;this.store =(KeyValueStore<String,Integer>) context.getStateStore("my-store");}@OverridepublicKeyValue<String,String>transform(String key,String value){// Custom transformation logic using the storeint currentCount = store.get(key)!=null? store.get(key):0;
        store.put(key, currentCount +1);returnnewKeyValue<>(key, currentCount +1+" occurrences");}@Overridepublicvoidclose(){// Close resources if needed}}// 使用 TransformerStreamsBuilder builder =newStreamsBuilder();
builder.stream("input-topic").transform(()->newMyTransformer(),"my-store").to("output-topic");

在这个例子中，自定义的

MyTransformer

类实现了

Transformer

接口，它在

init

方法中获取状态存储，并在

transform

方法中执行具体的转换逻辑，将每个键值对的值（计数）存储在状态存储中，并返回更新后的值。

总结

Kafka Streams 提供了一套丰富的 API，使得开发者能够便捷地处理 Kafka 中的数据流。基本概念如流、KStream、KTable、时间概念、窗口和状态管理构成了流处理的基础。而其 API 层面，高级 DSL 与低级 Processor API 结合，满足了从简单到复杂的不同应用场景的需求。通过熟练掌握这些概念和 API，开发者能够构建出高效、健壮的实时数据管道和流处理应用程序。Apache Kafka Streams 是一个用于构建实时流处理应用程序的客户端库，它直接构建在 Kafka 之上，允许开发者使用纯 Java 或 Scala 代码轻松处理 Kafka 中的数据流。以下是对 Kafka Streams 的基本概念及 API 的详解：

基本概念

**1. **流（Stream）

流是 Kafka Streams 最基础的抽象，它代表一个无限的、持续更新的数据集。每个流由一系列带有键值对的不可变记录组成，这些记录按照其在流中的位置（即偏移量）排序。流中的数据可以被重播，因为它持久存储在 Kafka 主题中，并且支持故障转移，使得流处理应用程序能够可靠地处理数据。

**2. **KStream 和 KTable

Kafka Streams 提供了两种核心的流处理抽象：

• KStream: 用于表示无界、持续更新的数据流，类似于传统的流处理概念。KStream 可以执行过滤、映射、聚合等操作，也可以与其他 KStream 或 KTable 进行连接（join）。
• KTable: 代表一个随时间变化的、键值对形式的表。KTable 适合存储具有唯一键的、随着时间推移可能会发生更新的数据。当新记录到达时，KTable 会更新其内部状态，反映最新的键值对。KTable 支持连接（join）和聚合操作，特别适用于实现基于事件时间的窗口化聚合。

**3. **时间概念

Kafka Streams 支持两种时间概念：

• 处理时间（Processing Time）: 当前系统时间，即流处理应用程序在处理记录时的实际时间。
• 事件时间（Event Time）: 记录中携带的时间戳，代表事件在其源头发生的时间。事件时间允许处理乱序事件和实现精确的基于时间窗口的聚合。

**4. **窗口（Window）

窗口是流处理中用来对无限数据流进行有限处理的重要概念。Kafka Streams 提供了时间窗口（如固定窗口、滑动窗口、会话窗口）和计数窗口，允许在特定时间段内或特定数量的记录范围内对数据进行聚合。

**5. **状态管理

Kafka Streams 具有内置的状态管理能力，允许应用程序维护有状态的处理逻辑。状态可以是局部的（每个实例有自己的状态副本）或全局的（所有实例共享状态）。状态存储在本地（通常使用 RocksDB），并通过 Kafka 的复制机制实现容错。

Kafka Streams API详解

Kafka Streams API 主要分为两个层次：

**1. **高级 DSL（Domain Specific Language）

提供了易于使用的高层接口，通过一系列操作符（如

map

,

filter

,

groupByKey

,

reduce

,

join

,

windowedBy

等）构建流处理逻辑。DSL 使开发者无需关心底层细节，可以快速编写简洁、声明式的流处理代码。

例如：

KStream<String,String> textLines = builder.stream("input-topic");KStream<String,Long> wordCounts = textLines
    .flatMapValues(line ->Arrays.asList(line.toLowerCase().split("\\W+"))).filter((key, word)->!word.isEmpty()).groupBy((key, word)-> word).count(Materialized.as("counts-store"));
wordCounts.to("output-topic",Produced.with(Serdes.String(),Serdes.Long()));

这段代码定义了一个从

input-topic

读取文本行，进行单词拆分、过滤、按单词分组并计数，最后将结果写入

output-topic

的流处理任务。

**2. **低级 Processor API

对于更复杂或定制化的场景，可以使用 Processor API 直接定义处理器（

Processor

、

Transformer

、

Punctuator

等）和处理器链。Processor API 提供了对流处理过程的细粒度控制，但要求开发者自行管理状态存储、定时器等。

例如，定义一个自定义

Transformer

：

classMyTransformerimplementsTransformer<String,String,KeyValue<String,String>>{ProcessorContext context;KeyValueStore<String,Integer> store;@Overridepublicvoidinit(ProcessorContext context){this.context = context;this.store =(KeyValueStore<String,Integer>) context.getStateStore("my-store");}@OverridepublicKeyValue<String,String>transform(String key,String value){// Custom transformation logic using the storeint currentCount = store.get(key)!=null? store.get(key):0;
        store.put(key, currentCount +1);returnnewKeyValue<>(key, currentCount +1+" occurrences");}@Overridepublicvoidclose(){// Close resources if needed}}// 使用 TransformerStreamsBuilder builder =newStreamsBuilder();
builder.stream("input-topic").transform(()->newMyTransformer(),"my-store").to("output-topic");

在这个例子中，自定义的

MyTransformer

类实现了

Transformer

接口，它在

init

方法中获取状态存储，并在

transform

方法中执行具体的转换逻辑，将每个键值对的值（计数）存储在状态存储中，并返回更新后的值。

总结

Kafka Streams 提供了一套丰富的 API，使得开发者能够便捷地处理 Kafka 中的数据流。基本概念如流、KStream、KTable、时间概念、窗口和状态管理构成了流处理的基础。而其 API 层面，高级 DSL 与低级 Processor API 结合，满足了从简单到复杂的不同应用场景的需求。通过熟练掌握这些概念和 API，开发者能够构建出高效、健壮的实时数据管道和流处理应用程序。