一、簡(jiǎn)介

1. 什么是kafka stream?

? Kafka Stream是Apache Kafka從0.10版本引入的一個(gè)新特性。它是一個(gè)用于處理和分析存儲(chǔ)在Kafka中的數(shù)據(jù)，并將得到的數(shù)據(jù)寫(xiě)回Kafka的客戶端程序庫(kù)。

2.kafka stream的特點(diǎn)

1. 依賴少.除kafka外，無(wú)其他依賴。
2. 提供了一個(gè)簡(jiǎn)單而輕量的jar包，可以很方便的嵌入到j(luò)ava程序中,方便打包與部署。
3. 基于Kafka的分區(qū)機(jī)制和Rebalance機(jī)制，實(shí)現(xiàn)水平擴(kuò)展和在線動(dòng)態(tài)調(diào)整并行度。
4. 提供記錄級(jí)的處理能力，從而實(shí)現(xiàn)毫秒級(jí)的低延遲。
5. 支持通過(guò)狀態(tài)存儲(chǔ)stateStore實(shí)現(xiàn)狀態(tài)操作以及支持基于事件時(shí)間的窗口操作。
6. 提供高級(jí)別(DSL)、低級(jí)別(processor)兩套操作API。

3.核心概念

3.1.拓?fù)?Topology)

拓?fù)錇閗afka stream處理的邏輯圖譜,由來(lái)源,一個(gè)個(gè)邏輯處理器節(jié)點(diǎn),流的流向以及流的輸出構(gòu)成.kafka stream提供兩種方式來(lái)構(gòu)建拓?fù)?

①kafka stream DSL(高級(jí)別)提供了常用的數(shù)據(jù)轉(zhuǎn)化操作,如:filter,map,count等.

②processor(低級(jí)別) 允許開(kāi)發(fā)者自己定義處理邏輯,以及基于狀態(tài)倉(cāng)庫(kù)(stateStore)做計(jì)算.

image

3.2.時(shí)間

kafka stream中時(shí)間的概念:

• 事件時(shí)間: 當(dāng)一個(gè)數(shù)據(jù)記錄發(fā)生的時(shí)間點(diǎn)，也就是數(shù)據(jù)被創(chuàng)建的時(shí)間。
• 處理時(shí)間: 數(shù)據(jù)記錄被流處理的時(shí)間,也即是數(shù)據(jù)被kafka stream消費(fèi)的時(shí)間.
• 攝取時(shí)間: 數(shù)據(jù)記錄被kafka broker存儲(chǔ)在topic分區(qū)的時(shí)間.

kafka stream在0.10后允許實(shí)現(xiàn)org.apache.kafka.streams.processor.TimestampExtractor接口,基于該接口,可根據(jù)業(yè)務(wù)需求自定義執(zhí)行不同的時(shí)間.

當(dāng)kafka stream處理完數(shù)據(jù)寫(xiě)回到kafka中時(shí),kafka stream將分配時(shí)間戳給新的消息.分配規(guī)則有上下文決定:

• 當(dāng)通過(guò)處理一些輸入記錄（例如，在process（）函數(shù)調(diào)用中觸發(fā)的context.forward（））生成新的輸出記錄時(shí)，輸出記錄時(shí)間戳直接從輸入記錄時(shí)間戳繼承。
• 當(dāng)通過(guò)周期性函數(shù)（如punctuate()）生成新的輸出記錄時(shí)。輸出記錄時(shí)間戳被定義為流任務(wù)的當(dāng)前內(nèi)部時(shí)間（通過(guò)context.timestamp()獲?。?。
• 對(duì)于聚合操作，生成的聚合更新的記錄時(shí)間戳將被最新到達(dá)的輸入記錄觸發(fā)更新。

3.3.狀態(tài)

一些流處理程序不需要狀態(tài),也就是每條消息處理獨(dú)立于其他的消息處理,如:過(guò)濾字符,文本打標(biāo)簽等.而另一些流處理程序是需要狀態(tài)的,如:網(wǎng)站PV量的計(jì)算等,kafka stream提供了狀態(tài)存儲(chǔ)功能,流處理程序可以用來(lái)存儲(chǔ)和查詢數(shù)據(jù).kafka stream默認(rèn)將數(shù)據(jù)存儲(chǔ)在本地RocksDB數(shù)據(jù)庫(kù)中.

3.4.分區(qū)(Partition)和任務(wù)(Task)

Kafka分區(qū)數(shù)據(jù)的消息層用于存儲(chǔ)和傳輸。Kafka Streams分區(qū)數(shù)據(jù)用于處理。基于kafka topic分區(qū)的并行性模型,kafka stream使用了分區(qū)和任務(wù)的概念.

Kafka Streams根據(jù)輸入流分區(qū)創(chuàng)建固定數(shù)量的Task，其中每個(gè)Task分配一個(gè)輸入流的topic. 分區(qū)對(duì)Task的分配不會(huì)改變，因此每個(gè)Task是應(yīng)用程序并行性的固定單位。然后，Task可以基于分配的分區(qū)實(shí)現(xiàn)自己的處理器拓?fù)?如果某個(gè)Stream的輸入Topic有多個(gè)(比如2個(gè)Topic，1個(gè)Partition數(shù)為4，另一個(gè)Partition數(shù)為3)，則總的Task數(shù)等于Partition數(shù)最多的那個(gè)Topic的Partition數(shù)。這是因?yàn)镵afka Stream使用了Consumer的Rebalance機(jī)制，每個(gè)Partition對(duì)應(yīng)一個(gè)Task。

并行模式下的分布:

image.png

3.5 數(shù)據(jù)抽象

1. KStream: data as record stream,KStream為一個(gè)insert隊(duì)列,新數(shù)據(jù)不斷增加進(jìn)來(lái).
2. KTable: data as change log stream,KTable為一個(gè)update隊(duì)列,新數(shù)據(jù)和已有數(shù)據(jù)有相同的key,則用新數(shù)據(jù)覆蓋原有原來(lái)的數(shù)據(jù).

流表二元性：

• 流作為表: 一個(gè)流可以認(rèn)為是一個(gè)表的變更日志，其中在流中的每個(gè)的數(shù)據(jù)記錄捕獲表的狀態(tài)變化。因此，流其實(shí)是一個(gè)偽裝的表，并且可以通過(guò)從開(kāi)始到結(jié)束重放變更日志來(lái)很容地重構(gòu)表。
• 表作為流: 表可以認(rèn)為是在流中的每個(gè)key的最新value的一個(gè)時(shí)間點(diǎn)的快照(流的數(shù)據(jù)記錄是k-v鍵值對(duì))。因此，表也可以認(rèn)為是偽裝的流，它可以通過(guò)對(duì)表中每個(gè)k-v進(jìn)行迭代而容易的轉(zhuǎn)換成流。

二、架構(gòu)

image.png

如圖所示，kafka stream支持接收多個(gè)topic中傳來(lái)的數(shù)據(jù)，并且通過(guò)kafka的rebalance機(jī)制，各個(gè)程序之間支持水平擴(kuò)展。kafka stream從kafka中獲取數(shù)據(jù)，并且內(nèi)置了consumer和producer。通過(guò)內(nèi)置的consumer接收到kafka中的數(shù)據(jù)，處理后再通過(guò)內(nèi)置的producer將數(shù)據(jù)返回到kafka中。

三、為什么選擇Kafka Stream?

1. 使用成本低

與Spark和Storm等流式處理框架相比，kafka stream提供的是一個(gè)基于kafka的流式處理類庫(kù)。且kafka stream作為流式處理類庫(kù)，直接提供具體的類和接口給開(kāi)發(fā)者，整個(gè)程序處理邏輯全都由開(kāi)發(fā)者自己控制，方便開(kāi)發(fā)和調(diào)試。

2.輕量易部署

由于kafka stream是作為類庫(kù)嵌入程序中，使得kafka stream打包部署非常方便。并且kafka stream利用了kafka 的分區(qū)機(jī)制和consumer的rebalance機(jī)制，使得kafka stream程序可以非常方便的進(jìn)行水平擴(kuò)展，并且可以在線動(dòng)態(tài)調(diào)整并行度。

3.性能

Kafka Stream的并行模型中，最小粒度為T(mén)ask，而每個(gè)Task包含一個(gè)特定子Topology的所有Processor，使得所有處理邏輯都在同一線程內(nèi)完成。這一特點(diǎn)跟Storm的Topology完全不一樣。Storm的Topology的每一個(gè)Task只包含一個(gè)Spout或Bolt的實(shí)例。因此Storm的一個(gè)Topology內(nèi)的不同Task之間需要通過(guò)網(wǎng)絡(luò)通信傳遞數(shù)據(jù)，而Kafka Stream的Task包含了完整的子Topology，所以Task之間不需要傳遞數(shù)據(jù)，也就不需要網(wǎng)絡(luò)通信。這一點(diǎn)降低了系統(tǒng)復(fù)雜度，也提高了處理效率。

四、缺點(diǎn)

• 暫不支持異步操作.所以在處理邏輯中避免使用高開(kāi)銷的操作,否則整個(gè)處理線程將會(huì)阻塞.
• 不支持像spark streaming那樣使用SQL完成實(shí)時(shí)的日志數(shù)據(jù)統(tǒng)計(jì).
• 數(shù)據(jù)來(lái)源單一,只支持kafka作為數(shù)據(jù)來(lái)源.

應(yīng)用示例

疑問(wèn)解答

1. Task和線程之間的關(guān)系

• kafka stream通過(guò)props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG,2);屬性可以設(shè)置并行的線程數(shù).
• Task的數(shù)量由Topic的分區(qū)數(shù)決定,取監(jiān)聽(tīng)的topic中最大的分區(qū)數(shù)作為T(mén)ask的數(shù)量,Task和Thread之間的分配由線程數(shù)決定.若有4個(gè)Task,但是只有一個(gè)Thread,則4個(gè)Task位于同一線程中串行.若有4個(gè)Task,和4個(gè)Thread,則每個(gè)Task獨(dú)享一個(gè)線程,并行處理.若Task數(shù)大于Thread數(shù),則有kafka stream自行做分配.若Task數(shù)小于線程數(shù),則會(huì)出現(xiàn)某些線程不能執(zhí)行Task的情況.

2. 當(dāng)某一實(shí)例處理數(shù)據(jù)時(shí)宕機(jī)了,數(shù)據(jù)是否會(huì)丟失

   分區(qū)與任務(wù)的分配永遠(yuǎn)不改變,當(dāng)應(yīng)用實(shí)例執(zhí)行任務(wù)失敗時(shí),則其被分配的任務(wù)將自動(dòng)在其他實(shí)例中被創(chuàng)建,并從相同的流分區(qū)重新消費(fèi).