Maxkit: Kafka

2016/07/18

Kafka

Kafka 是一種 distributed, partitioned, replicated commit log service，核心概念是 messaging system。

basic messaging terminology

topic: 訊息的分類
producer: 發佈訊息到 topics
consumer: 註冊 topics 並持續處理 published messages
broker: 由一到多個 broker servers 可組成 cluster

Kafka cluster 會針對每一個 topic 維護一個類似以下圖形的 partitioned log。

每一個 partition 都是 ordered, immutable sequences of messages，並會持續附加到 commit log，訊息在 partitions 中會被 assigned 一個 sequential id number 稱為 offset，代表這個訊息在 partition 中的位置。

partition 可讓 log size 超過單一 server 的限制，雖然每一個 partition 都受到 server 的限制，但組合多個 partitions 的 topic 就可以處理任意數量的資料。partition 的另一個優點是 parallelism 平行處理。

在 Kafka cluster 中，為了處理 fault tolerance，每個 partition 可設定複製到多個 servers。這些 partitions 會設定讓某個 server 的 partition 成為 "leader" 身份，所有讀寫的 request 都會由他處理，其他的 server 是 "followers" 身份，他們只會複製 partitions，當 "leader" 消失，所有 followers 中會找出一個新的 "leader"。每個 server 都可能是某些 partitions 的 leader，用以讓這個 cluster well-balanced。

offset 是由 consumer 控制，通常 consumer 會在讀取訊息時，自己(linearly)增加 offset，但實際上，consumer 可以自由以任何順序讀取並消化訊息。唯一一個會永久保存的資訊是，每一個 consumer 的 offset 位置。

不管訊息有沒有被消化，Kafka cluster 將會保留所有發佈的訊息一段時間，時間長短可設定調整，舉例來說，log retention 應用會設定為 2天。

Producers, Consumers

Producer 發佈訊息到 topic 時，是以 round-robin 的方式 load balanced，也可以依照某個 partition function 來決定。

Consumers 有兩種模式：queuing 與 publish-subscribe，queuing 模式中，每一個訊息只會丟給一堆 consumers 中的某一個來處理，publish-subscribe 模式中，訊息會 broadcast 到所有 consumers。

Kafka 會用 consumer group name 來標記 consumer group，每一個訊息是發送給 consumer group 中的某一個 consumer instance，consumer instances 可以在不同的 processes 或是不同機器上。

通常為了 fault tolerance 的緣故，topics 有 small number of consumer groups。

2 server Kafka cluster, 4 partitions(P0~P3), 2 consumer groups, 2 consumer instances in consumer group A

Kafka 可確保訊息的順序，可在 a pool of consumer processes 中提供 ordering quarantees 以及 load balancing。有個條件限制是，在 consumer group 中，consumer instance 的數量不能超過 partitions 的數量。

Kafka 只能確保在一個 partition 中的訊息順序，如果是 topic 的多個 partitions 則無法保證順序。如果應用是需要 topic 裡面訊息的順序，就只能讓 topic 只有一個 partition，這表示每一個 consumer group 只能有一個 consumer process。

Guarantees

Kafka 保證以下事項

producer 發送給一個特殊的 topic partition 的訊息，會根據發送的順序來排序。先發送的訊息會有比較低的 offset，比較早出現在 log 中。
consumer 讀取訊息的順序會根據儲存在 log 的順序來決定。
對於一個有 replication factor N 的 topic 來說，可容許 N-1 個 server failures，不會遺失任何已經 commit 到 log 中的訊息。

Use Cases, 幾個 Kafka 的應用情境

Messaging

Kafka 跟其他 messaging system 的差別是：better throughput, built-in partitioning, replication, and fault-tolerance 通常 messaging system 是 low-throughput 的，但實際上卻需要 low end-to-end latency，也需要有 stong durability 的功能。
Website Activity Tracking

利用 real-time publish-subscribe 方式，建立 user activity traking pipeline。可提供 real-time processing, real-time monitoring, and loading into Hadoop or offline data warehousing systems for offline processing and reporting 這些功能。
Metrics

operational monitoring data，在 distributed application 中整合統計資料。
Log Aggregation

Log aggregation 用來收集 server 的 physical log files，集中放在 file system 裡面。Kafka 將 log 轉換成 a steam of messages，建立 abstraction of log system
Stream Processing

處理 stage-wise processing of data，由 raw data 經過處理與轉換，產生新的資料，例如由 RSS 經過 crawl article content，然後發佈到 topic: articles，然後進行 normalize 或 deduplicate content，最後推薦給使用者建議閱讀的文章。Storm 與 Samza 是處理這種轉換的常用 framework。
Event Sourcing

event sourcing 是一種 application design 的形式，state change 會依照時間順序記錄起來。
Commit Log

Kafka 可作為 distributed system 的 external commit-log，可在 nodes 之間複製資料，並在 failed nodes 用來 restore data。Apache BookKeeper 就是這種應用的類似的 framework。