1.概述
Kafka Streams 是一個用來處理流式數據的庫，屬於Java類庫，它並不是一個流處理框架，和Storm，Spark Streaming這類流處理框架是明顯不一樣的。那這樣一個庫是做什么的，能應用到哪些場合，如何使用。筆者今天就給大家來一一剖析這些內容。
2.內容
首先，我們研究這樣一個庫，需要知道它是做什么的。Kafka Streams是一個用來構建流處理應用的庫，和Java的那些內置庫一樣，以一種分布式的容錯方式來處理一些事情。當前，業界用於流處理的計算框架包含有：Flink，Spark，Storm等等。Kafka Streams處理完后的結果可以回寫到Topic中，也可以外接其他系統進行落地。包含以下特性：
- 事件區分：記錄數據發生的時刻
- 時間處理：記錄數據被流處理應用開始處理的時刻，如記錄被消費的時刻
- 開窗
- 狀態管理：本身應用不需要管理狀態，如若需要處理復雜的流處理應用（分組，聚合，連接等）
Kafka Streams使用是很簡單的，這一點通過閱讀官方的示例代碼就能發現，另外它利用Kafka的並發模型來完成負載均衡。
2.1 優勢
在Kafka集群上，能夠很便捷的使用，亮點如下圖所示：
- 能夠設計一些輕量級的Client類庫，和現有的Java程序整合
- 不需要額外的Kafka集群，利用現有的Kafka集群的分區實現水平擴展
- 容錯率，高可用性
- 多平台部署，支持Mac，Linux和Windows系統
- 權限安全控制
2.2 Sample
Kafka Streams是直接構建與Kafka的基礎之上的，沒有了額外的流處理集群，Table和一些有狀態的處理完全整合到了流處理本身。其核心代碼非常的簡介。簡而言之，就和你寫Consumer或Producer一樣，但是Kafka Streams更加的簡潔。
2.3 屬性
| 名稱 | 描述 | 類型 | 默認值 | 級別 |
| application.id | 流處理標識，對應一個應用需要保持一致，用作消費的group.id | string | 高 | |
| bootstrap.servers | 用來發現Kafka的集群節點，不需要配置所有的Broker | list | 高 | |
| replication.factor | 復制因子 | int | 1 | 高 |
| state.dir | 本地狀態存儲目錄 | string | /tmp/kafka-streams | 高 |
| cache.max.bytes.buffering | 所有線程的最大緩沖內存 | long | 10485760 | 中 |
| client.id | 客戶端邏輯名稱，用於標識請求位置 | string | "" | 中 |
| default.key.serde | 對Key序列化或反序列化類，實現於Serde接口 | class | org.apache.kafka.common.serialization.Serdes$ByteArraySerde | 中 |
| default.value.serde | 對Value序列化或反序列化類，實現與Serde接口 | class | org.apache.kafka.common.serialization.Serdes$ByteArraySerde | 中 |
| ... | ... | ... | ... | ... |
這里只是列舉了部分Kafka Streams的屬性值，更多的詳情可參考Kafka Streams Configs。
3.示例
下面，我們可以通過一個示例代碼，來熟悉Kafka Streams的運行流程，如下所示：
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KStreamBuilder;
import org.apache.kafka.streams.kstream.KTable;
import java.util.Arrays;
import java.util.Properties;
public class WordCountApplication {
public static void main(final String[] args) throws Exception {
Properties config = new Properties();
config.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount_topic_appid");
config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka1:9092,kafka2:9092,kafka3:9092");
config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
KStreamBuilder builder = new KStreamBuilder();
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("Counts");
wordCounts.to(Serdes.String(), Serdes.Long(), "WordsWithCountsTopic");
KafkaStreams streams = new KafkaStreams(builder, config);
streams.start();
}
}
從代碼中，我們可以看出Kafka Streams為上層流定義了兩種基本抽象：
- KStream：可以從一個或者多個Topic源來創建
- KTable：從一個Topic源來創建
這兩者的區別是，前者比較像傳統意義上的流，可以把每一個K/V看成獨立的，后者的思想更加接近與Map的概念。同一個Key輸入多次，后者是會覆蓋前者的。而且，KStream和KTable都提供了一系列的轉換操作，每個操作可以產生一個或者多個KStream和KTable對象，所有這些轉換的方法連接在一起，就形成了一個復雜的Topology。由於KStream和KTable是強類型，這些轉換都被定義為通用函數，這樣在使用的時候讓用戶指定輸入和輸出數據類型。
另外，無狀態的轉換不依賴於處理的狀態，因此不需要狀態倉庫。有狀態的轉換則需要進行存儲相應的狀態用於處理和生成結果。例如，在進行聚合操作的時候，一個窗口狀態用於保存當前預定義收到的值，然后轉換獲取累計的值，再做計算。
在處理完后，對於結果集用戶可以持續的將結果回寫到Topic，也可以通過KStream.to() 或者 KTable.to() 方法來實現。
4.總結
通過對Kafka Streams的研究，它的優勢可以總結為以下幾點。首先，它提供了輕量級並且易用的API來有效的降低流數據的開發成本，之前要實現這類處理，需要使用Spark Streaming，Storm，Flink，或者自己編寫Consumer。其次，它開發的應用程序可以支持在YARN，Mesos這類資源調度中，使用方式靈活。而對於異步操作，不是很友好，需要謹慎處理；另外，對SQL語法的支持有限，需要額外開發。
5.結束語
這篇博客就和大家分享到這里，如果大家在研究學習的過程當中有什么問題，可以加群進行討論或發送郵件給我，我會盡我所能為您解答，與君共勉。