QuickStart

快速啟動你的 Kafka

一開始是沒有很了解 Kafka 是什麼、怎麼運行，所以就先 run 起來試試看 ;D

Kafka Reference

Reference：
What is Apache Kafka?
Kafka QuickStart

以下翻譯 QuickStart ，並實際運行看看：

從下載、啟動、收發事件，到 Kafka Connect 與 Kafka Streams，八個步驟帶你在本機跑起一套完整的 Kafka 環境

步驟 1：取得 Kafka

下載最新的 Kafka 版本並解壓縮：

$ tar -xzf kafka_2.13-4.3.1.tgz
$ cd kafka_2.13-4.3.1

步驟 2：啟動 Kafka 環境

注意： 您的本機環境必須安裝 Java 17 以上版本。

Kafka 可以透過本機腳本與下載的檔案執行，也可以使用 Docker 映像檔執行。兩種方式最後都會得到一個在 9092 埠監聽的 broker。

使用下載的檔案

產生叢集 UUID：

$ KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"

格式化日誌目錄：

$ bin/kafka-storage.sh format --standalone -t $KAFKA_CLUSTER_ID -c config/server.properties

啟動 Kafka 伺服器：

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

Kafka 伺服器成功啟動後，您就有一個可以使用的基本 Kafka 環境了。

使用基於 JVM 的 Apache Kafka Docker 映像檔

取得 Docker 映像檔：

$ docker pull apache/kafka:4.3.1

啟動 Kafka Docker 容器：

$ docker run -p 9092:9092 apache/kafka:4.3.1

使用基於 GraalVM 的原生 Apache Kafka Docker 映像檔

取得 Docker 映像檔：

$ docker pull apache/kafka-native:4.3.1

啟動 Kafka Docker 容器：

$ docker run -p 9092:9092 apache/kafka-native:4.3.1

步驟 3：建立主題來儲存事件

Kafka 是一個分散式事件串流平台，可讓您在多台機器上讀取、寫入、儲存及處理事件（在文件中也稱為紀錄或訊息）。

事件的例子包括付款交易、手機的地理位置更新、出貨訂單、物聯網裝置或醫療設備的感測器量測值等等。這些事件會被組織並儲存在主題（topic）中。非常簡化地說，主題就像檔案系統中的資料夾，而事件就是資料夾裡的檔案。

因此，在寫入第一批事件之前，您必須先建立一個主題。開啟另一個終端機工作階段並執行：

$ bin/kafka-topics.sh --create --topic quickstart-events --bootstrap-server localhost:9092

Kafka 的所有命令列工具都有額外的選項：不帶任何參數執行 kafka-topics.sh 即可顯示使用說明。例如，它也可以顯示新主題的分割區數量等詳細資訊：

$ bin/kafka-topics.sh --describe --topic quickstart-events --bootstrap-server localhost:9092
Topic: quickstart-events        TopicId: NPmZHyhbR9y00wMglMH2sg PartitionCount: 1       ReplicationFactor: 1	Configs:
Topic: quickstart-events Partition: 0    Leader: 0   Replicas: 0 Isr: 0

小補充：如果我的 Leader 是 1，不是 0，代表有問題嗎？

Leader 不是數量，而是「負責這個分割區的 broker 的 ID」。

上面的範例輸出來自 broker ID 為 0 的環境；

Kafka 4.x 內附的 config/server.properties 預設為 node.id=1，使用 apache/kafka Docker 映像檔時預設也是 1，所以看到的通常會是：

Topic: quickstart-events Partition: 0    Leader: 1   Replicas: 1 Isr: 1

這是正常的。三個欄位的意思：

• Leader：處理這個分割區讀寫的 broker ID。
• Replicas：存放這個分割區複本的 broker ID（單機只有一份）。
• Isr：目前與 leader 保持同步的複本所在的 broker ID。

可以用 grep node.id config/server.properties 確認自己的設定。只要三個欄位的數字一致，就代表單節點環境運作正常。

步驟 4：將一些事件寫入主題

Kafka 用戶端透過網路與 Kafka broker 通訊，以寫入（或讀取）事件。broker 收到事件後，會以持久且具容錯能力的方式儲存，保存多久都可以，甚至永久保存。

執行主控台生產者（console producer）用戶端，將幾個事件寫入您的主題。預設情況下，您輸入的每一行都會成為一個獨立的事件寫入主題。

$ bin/kafka-console-producer.sh --topic quickstart-events --bootstrap-server localhost:9092
>This is my first event: kaka1
>This is my second event: kaka2

隨時可以按 Ctrl-C 停止生產者用戶端。

步驟 5：讀取事件

開啟另一個終端機工作階段，執行主控台消費者（console consumer）用戶端，讀取剛才建立的事件：

$ bin/kafka-console-consumer.sh --topic quickstart-events --from-beginning --bootstrap-server localhost:9092
This is my first event: kaka1
This is my second event: kaka2

隨時可以按 Ctrl-C 停止消費者用戶端。

歡迎自由實驗：例如，切換回生產者終端機（上一步）寫入更多事件，看看這些事件如何立即出現在消費者終端機中。

由於事件會持久地儲存在 Kafka 中，因此可以被任意多個消費者讀取任意多次。您可以輕鬆驗證這一點：再開一個終端機工作階段，重新執行上一個指令即可。

步驟 6：使用 Kafka Connect 以事件串流匯入／匯出資料

我們可能在關聯式資料庫或傳統訊息系統等既有系統中有大量資料，而且有許多應用程式已經在使用這些系統。Kafka Connect 可讓您持續地將資料從外部系統匯入 Kafka，反之亦然。

它是一個可擴充的工具，負責執行連接器（connector），連接器實作了與外部系統互動的自訂邏輯。因此，將既有系統與 Kafka 整合非常容易。為了讓這個過程更簡單，現成可用的連接器多達數百種。

在快速入門中，我們將以簡單的連接器執行 Kafka Connect：將資料從檔案匯入 Kafka 主題，以及將資料從 Kafka 主題匯出到檔案。

首先，請確認已將 connect-file-4.3.1.jar 加入 Connect worker 設定中的 plugin.path 屬性。在本快速入門中，我們使用相對路徑，並將連接器套件視為 uber jar；當指令在安裝目錄中執行時，這種方式是可行的。不過在正式環境部署時，使用絕對路徑總是比較好。關於此設定的詳細說明，請參閱 plugin.path。

編輯 config/connect-standalone.properties 檔案，新增或修改 plugin.path 設定使其符合以下內容，然後儲存檔案：

$ echo "plugin.path=libs/connect-file-4.3.1.jar" >> config/connect-standalone.properties

接著，先建立一些測試用的初始資料：

$ echo -e "foo\nbar" > test.txt

或在 Windows 上：

$ echo foo > test.txt
$ echo bar >> test.txt

接下來，我們將以獨立模式（standalone mode）啟動兩個連接器，也就是說它們會在單一、本機、專用的程序中執行。我們提供三個設定檔作為參數。第一個一律是 Kafka Connect 程序的設定，包含要連線的 Kafka broker 以及資料的序列化格式等通用設定。其餘的設定檔各自指定一個要建立的連接器，內容包含唯一的連接器名稱、要實例化的連接器類別，以及連接器所需的其他設定。

$ bin/connect-standalone.sh config/connect-standalone.properties config/connect-file-source.properties config/connect-file-sink.properties

這些 Kafka 內附的範例設定檔會使用先前啟動的預設本機叢集設定，並建立兩個連接器：第一個是來源（source）連接器，從輸入檔案讀取每一行並寫入 Kafka 主題；第二個是匯出（sink）連接器，從 Kafka 主題讀取訊息並將每則訊息寫成輸出檔案中的一行。

啟動期間會看到許多日誌訊息，其中包括表示連接器正在實例化的訊息。Kafka Connect 程序啟動後，來源連接器會開始從 test.txt 讀取每一行並寫入主題 connect-test，而匯出連接器會開始從 connect-test 讀取訊息並寫入 test.sink.txt。檢查輸出檔案內容即可驗證資料已走完整條管線：

$ more test.sink.txt
foo
bar

資料儲存在 Kafka 主題 connect-test 中，所以也可以執行主控台消費者查看主題中的資料（或使用自訂的消費者程式碼處理）：

$ bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic connect-test --from-beginning
{"schema":{"type":"string","optional":false},"payload":"foo"}
{"schema":{"type":"string","optional":false},"payload":"bar"}
…

連接器會持續處理資料，因此可以在檔案中新增資料，看著它流經整條管線：

$ echo "Another line" >> test.txt

應該會看到這一行出現在主控台消費者的輸出以及匯出檔案中。

步驟 7：使用 Kafka Streams 處理事件

當資料以事件的形式儲存在 Kafka 後，可以使用適用於 Java／Scala 的 Kafka Streams 用戶端函式庫來處理資料。它可讓我們實作關鍵任務型的即時應用程式與微服務，其輸入和／或輸出資料都儲存在 Kafka 主題中。

Kafka Streams 結合了在用戶端撰寫與部署標準 Java、Scala 應用程式的簡便性，以及 Kafka 伺服器端叢集技術的優勢，讓應用程式具備高度可擴展性、彈性、容錯能力且為分散式。此函式庫支援「恰好一次」（exactly-once）處理、有狀態操作與聚合、視窗化（windowing）、聯結（join）、基於事件時間的處理等功能。

先初步體驗一下，以下是熱門 WordCount 演算法的實作方式：

KStream<String, String> textLines = builder.stream("quickstart-events");

KTable<String, Long> wordCounts = textLines
            .flatMapValues(line -> Arrays.asList(line.toLowerCase().split(" ")))
            .groupBy((keyIgnored, word) -> word)
            .count();

wordCounts.toStream().to("output-topic", Produced.with(Serdes.String(), Serdes.Long()));

Kafka Streams 示範與應用程式開發教學，會從頭到尾示範如何撰寫並執行這樣的串流應用程式。

步驟 8：關閉 Kafka 環境

剛才完成了本快速入門，可以關閉 Kafka 環境，或繼續自由探索。

• 如果還沒停止，請按 Ctrl-C 停止生產者與消費者用戶端。
• 按 Ctrl-C 停止 Kafka broker。

如果也想刪除本機 Kafka 環境中的所有資料，包括過程中建立的所有事件，請執行以下指令（此動作無法復原）：

$ rm -rf /tmp/kafka-logs /tmp/kraft-combined-logs

內容譯自 Apache Kafka 官方快速入門文件（4.3.1 版），流程圖為輔助說明。