Kafka Connect內部原理

今天主要講在大規模數據情況下，Kafka如何解決實時的數據集成問題。主要有以下四個方面：

1. Traditional ETL

2. Data Integration

3. Kafka Connect

4. Group Management

1. Traditional ETL

回顧歷史，三四十年前，大部分的應用是比較簡單的架構（如下圖），主要數據存儲在關係型資料庫里。關係型資料庫里的數據每過一段時間就放在數據倉庫里做離線分析，以此找到數據中的一些模式。這時候數據源只有一個。

現在有很多新的系統出現，如下圖：

這些存儲系統有兩類，一種是primary storage（主存儲，比如NoSQL），它是presistent的。現在數據源會有很多個，關係型和非關係型資料庫的數據都需要導到數據倉庫中。另一種是secondary storage（二級存儲），數據一般是從primary storage做變化後得到，即使數據丟失，也可以從primary storage中恢復,對presistent要求沒有那麼高。比如elastic search，它適合對全文檢索和實時性要求高的場所。

隨著系統種類的增多，系統之間傳遞數據的實時性和連續性變得非常強。因為所有的系統服務於我們的應用。我們希望數據在主存儲里一旦更改，就能在二級存儲里改變。

一旦存儲系統變多，在不同系統間傳遞數據變成非常複雜的事情。下圖是簡化版：

一份數據因為用處不同，需要拷貝到很多地方去。如果圖中每條線都用一個專門的工具處理，耗時且難以實現。

2. Data Integration

Data integration就是將數據放到應該去的地方。

首先，先介紹不應該有的設計，如下圖：

1）one-off tools

One-off tools如之前圖中所示，每條線用一個專門的工具來做。

2）kitchen sink tools

Kitchen sink tools指特別通用的工具，能在所有系統間導數據。它的問題是因為太通用，對source和destination的假設太少，無法確保reliable。

3）stream processing frameworks

它的側重點在transformation，不關注如何支持在不同系統間拷貝數據。因此需要自己做很多工作。

其中1）和2）是兩個極端，如下圖左側為one-off tools的情況，右側ET&L（Extract-Transform-Load）為kitchen sink tools情況，指從資料庫中找到數據，做一些變換，然後load到數據倉庫中。

如何在這兩者之間做個平衡，是kafka最初設計的動力。

Kafka是個place holder，假設所有的東西都往kafka發數據或取數據，那麼所有系統都和kafka連接，不會出現強耦合。這樣對kafka會有很多要求。

Kafka是一個分散式系統，它有很多brokers。

Kafka以log的方式管理數據，順序讀寫並且寫只能在log尾部。

Kafka和pub-sub系統的聯繫在於source不斷地寫消息，sub不斷的讀消息。

為了使kafka能scale，引入topic的概念。Topic相當於partitioned log。一個topic可以有很多個partition。Kafka只能保證在一個partition里有順序。

Kafka還支持scalable consumption。它是指有很多個consumer，成為consumer group，可以訂閱很多不同的topic。比如consumer group A，訂閱了partition 0，1，2。因為group里有兩個consumer，不同的partition會分配到不同consumer。假設增加一個新的consumer，屬於consumer goup A，那麼partition 0或者1就會放到新的consumer里，產生新的平均分配。可以把group想像成系統。

Kafka不決定如何consume，consumer自己決定何時，如何consume。Kafka里的數據是持久化的，每個數據都存在伺服器硬碟里。

Kafka里的consumer支持fault tolerant。如果一個consumer掛到，kafka會把任務放到其他consumer上。

3. Kafka Connect

前面所說的consumer和producer都是develop的工具。如果想用它們做data pipeline或者ETL的工作，要寫大量的數據。同時還要對data pipeline做管理和監控。Kafka connect解決了這個廣泛的需求。

它的目標是：

1）它著重於數據的copying，而不是transiformation。

2）它解決fault tolerance，scale up，scale down，monitor和mangement的工作。

3）所有connect job都會有統一的方法來monitor和management。

4）所有的job都是並行運行。

5）可以用單機模式運行你的任務。

Kafka connect有兩個概念，一個source，另一個是sink。source是把數據從一個系統拷貝到kafka里，sink是從kafka拷貝到另一個系統里。如下圖：

Kafka的數據模型是它是一個partitioned的stream。如果把一個partition看做一個表格，並且每個partition有一個primary key或者timestamp，可以根據timestamp來order表中數據，就可以把它看做一個stream了。

下圖是一個具體的例子，id相當於timestamp：

以這個角度看待資料庫，資料庫就變成了流數據。

另一個模型是一個connector有很多個task。我們可以把每個partition放在一個thread里運行，但這不是一個很好的模型，因為parittion的數量也許會很大，因此引入task。多個partition可以放在一個task里，由一個thread執行。這樣資源的邏輯區分和物理劃分可以更加匹配。

Kafka有兩種運行模式，第一種是standalone mode，如下圖：

這種是單機模式，另一種是分散式模式，如下圖：

一台或很多台機器里有多個worker進程，worker進程有很多connector和task。connector負責獲得所要連接的系統的信息，並不進行數據拷貝。task進行數據拷貝。這個模式也是fault torerant，如果worker 4掛了，它的connector和task將交由worker 3進行。

4. Group Management

以前的group management是由zookeeper來實現，但是這樣rebalance的負擔太重。為了解決scalable的問題，不再使用zookeeper，讓每個broker負責一些group的管理，client端完全不需要依賴zookeeper，開發管理變得更加簡單。這個group management有兩個階段：第一階段發現group里都有誰，第二階段讓每個memeber狀態同步。下圖是個例子：

Coordinator可以理解成kafka的broker，member可以是每個consumer group里的group或者connect里的worker，member通過id來確認group。member通過JoinGroupRequest找到coordinator。coordinator等一段時間（session time out，default為30秒）來接收各種request。之後coordinator會發response給member。

Response會包括member里leader信息。coordinator收到的第一個請求的發送方就是leader。同時coordinator會把所有member的信息發給每個member。

第二階段，leader決定分配的任務。用leader分配任務是因為，coordinator不知道member之間使用什麼策略來分配任務。

總的來說，group management的工作分類如下：

Kafka connect保證at least one delivery：

最後，簡單介紹以下如何做到exactly once：

其中WAL涉及到zombie writer：

圖中C為coordinator，T為task。每個task都有一個consumer instance。對一個connector來說，它所有task的consumer instance都在一個consumer group里。假設有個task特別慢，造成consumer和coordinator無法通信。Coordinator長時間無法檢測到心跳會把這個consumer踢出consumer group，觸發rebalance。這個task的任務會被放在其他task上。在commit（把臨時文件改為永久文件）之前，數據都在臨時文件里。如果T1被踢出group前的offset為100-200，T2為100-300，並且都還沒有commit，那麼可能會出現T1之後再次激活，T1和T2都要commit，導致數據的重複。Read head log能避免這種情況的發生。在任何commit之前，都要先寫到read head log里，把這個操作放到HDFS里的一個文件里，然後flush到所有copy里。這樣如果T1即使再次激活也不能再寫文件，T2執行所有的文件（執行時檢查文件是否被寫），同時更新HDFS文件。

Reference

1. About the Confluent Platform

2. Confluent Blog

3. Announcing Kafka Connect: Building large-scale low-latency data pipelines

4. How to Build a Scalable ETL Pipeline with Kafka Connect

5. Kafka Client-side Assignment Proposal

本文作者：Shaoke Xu，查看完整視頻：http://www.bittiger.io/classes

更多內容，請訪問：BitTiger.io， 掃描下面二維碼，關注微信公眾賬號「論碼農的自我修養」