Kafka 簡介

Apache Kafka 是一個分散式事件流平台，提供高效的消息發布與消息訂閱服務，具有高吞吐、可擴展、永久存儲及高可用性等特點，被廣泛地應用在大數據的 data pipeline 當中。

架構

圖片來源: Kafka Broker, Kafka Topic, Consumer and Record Flow in Kafka

• 生產者 (Producer)
  發佈消息到一個或多個主題上，並指定分區。
• 消費者 (Consumer)
  訂閱主題以獲取消息，可以訂閱整個主題或是只訂閱特定分區，或是透過消費者群組 (Consumer group) 讓多個消費者共同處理主題。
• 伺服器 (Broker)
  集群中的每個伺服器都稱為 Broker，是獨立的消息存儲與處理單元，透過 ZooKeeper、KRaft 等工具來管理集群的元數據 (metadata) 與協調工作。
• 主題 (Topic)
  消息的類別，是消息分發的基本單位。
• 分區 (Partition)
  一個主題可以有多個分區，每個分區都是一個消息的存儲單位，一個消息會被存儲在一個或多個分區之中，每個分區都可以獨立處理消息進而實現平行處理。

Kafka vs Flink

• Kafka 主要是消息佇列系統，不太會關注數據的實時處理。
• Flink 是一個流處理框架，關注數據的實時處理。

兩者的用途不同，但能夠很好的配合使用，Flink 可以做為 Consumer 訂閱 Kafka 主題，同時也可做為 Producer 將結果發佈到 Kafka 主題上。

Kafka 安裝與設置

至官網下載二進制檔或使用下列指令下載 (本次使用的版本是 3.5.0)：

• 安裝

  $ mkdir kafka
  $ cd kafka
  $ wget https://dlcdn.apache.org/kafka/3.5.0/kafka_2.13-3.5.0.tgz
  $ tar -zxvf kafka_2.13-3.5.0.tgz
  $ rm kafka_2.13-3.5.0.tgz
  $ cd kafka_2.13-3.5.0
  

• 設置環境變數

  • 編輯 ~/.bashrc

    $ vim ~/.bashrc
    

  • 將下方變數加入 ~/.bashrc

    export KAFKA_HOME=~/kafka/kafka_2.13-3.5.0
    

  • 重新讀取 ~/.bashrc

    $ source ~/.bashrc
    

運行 Kafka

Apache Kafka can be started using ZooKeeper or KRaft.

ZooKeeper

ZooKeeper 是一個分散式協調服務，可以用來分享配置信息、元數據等，在較早的版本中，Kafka 必須透過 ZooKeeper 來協調管理各個主分區的元數據與狀態。

• 啟動 ZooKeeper

  $ $KAFKA_HOME/bin/zookeeper-server-start.sh config/zookeeper.properties
  

• (開新 terminal) 啟動 Kafka 伺服器

  $ $KAFKA_HOME/bin/kafka-server-start.sh config/server.properties
  

KRaft

Kafka 在 2.8.0 後提出了內建的共識機制 KRaft，目的是移除 對於 ZooKeeper 的依賴，具體機制可參考：脱离ZooKeeper依赖的Kafka Controller Quorum（KRaft）机制浅析

• 初始化 KRaft 集群

  $ KAFKA_CLUSTER_ID="$($KAFKA_HOME/bin/kafka-storage.sh random-uuid)"
  $ $KAFKA_HOME/bin/kafka-storage.sh format -t $KAFKA_CLUSTER_ID -c $KAFKA_HOME/config/kraft/server.properties
  

• 啟動 Kafka 伺服器

  $ $KAFKA_HOME/bin/kafka-server-start.sh $KAFKA_HOME/config/kraft/server.properties
  

基本操作

建立主題 (Create Topic)

$ $KAFKA_HOME/bin/kafka-topics.sh --create --topic <topic_name> --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1

• bootstrap-server: 用來指定 Kafka 伺服器的位置與通訊埠
• partitions: 用來設定主題的分區數量
• replication-factor: 用來設定複製因子 (每個分區的複本數)

查看主題 (Topic List)

$ $KAFKA_HOME/bin/kafka-topics.sh --list --bootstrap-server localhost:9092

刪除主題 (Delete Topic)

$ $ $KAFKA_HOME/bin/kafka-topics.sh --delete --topic <topic_name> --bootstrap-server loca
lhost:9092

發佈消息 (Produce Event to Topic)

$ $KAFKA_HOME/bin/kafka-console-producer.sh --topic <topic_name> --bootstrap-server localhost:9092

接收消息 (Read Event from Topic)

$ $KAFKA_HOME/bin/kafka-console-consumer.sh --topic <topic_name> --bootstrap-server localhost:9092

預告

明天會介紹如何在 Flink 中使用 Apache Kafka Connector 來收發數據流。

參考資料

APACHE KAFKA QUICKSTART