JSON 文件數據導入 Kafka 主題
1. 概述
Apache Kafka 是一個開源、容錯且高度可擴展的流媒體平台。它遵循發布-訂閱架構來實時傳輸數據。通過將數據放入隊列，我們可以以非常低的延遲處理處理大量海量數據。有時，我們需要將JSON數據類型發送到Kafka主題進行數據處理和分析。
在本教程中，我們將學習如何將 JSON 數據流式傳輸到 Kafka 主題中。此外，我們還將了解如何為 JSON 數據配置 Kafka 生產者和消費者。
2. JSON數據在Kafka中的重要性
從架構上來說，Kafka 在其係統中支持消息流。因此，我們也可以將JSON數據發送到Kafka服務器。如今，在現代應用系統中，每個應用程序主要只處理 JSON，因此以 JSON 格式進行通信變得非常重要。通過發送 JSON 格式的數據，有利於實時跟踪用戶及其在網站和應用程序上的行為。
將 JSON 類型的數據傳輸到 Kafka 服務器有助於實時數據分析。它促進了事件驅動的架構，其中每個微服務訂閱其相關主題並實時提供更改。使用 Kafka 主題和 JSON 格式，可以輕鬆交付 IOT 數據、微服務之間通信以及聚合指標。
3.卡夫卡設置
要將 JSON 流式傳輸到 Kafka 服務器，我們需要首先設置 Kafka 代理和 Zookeeper。我們可以按照本教程設置一個成熟的 Kafka 服務器。現在，讓我們檢查一下創建 Kafka 主題baeldung命令，我們將在該主題上生成和使用 JSON 數據：
$ docker-compose exec kafka kafka-topics.sh --create --topic baeldung
--partitions 1 --replication-factor 1 --bootstrap-server kafka:9092上面的命令創建一個複制因子為1的 Kafka 主題baeldung 。在這裡，我們創建了一個只有1複製因子的 Kafka 主題，因為它僅用於演示目的。在實際場景中，我們可能需要多複制因子，因為它有助於系統故障轉移。此外，它還提供數據的高可用性和可靠性。
4. 產生數據
Kafka生產者是整個Kafka生態系統中最基本的組成部分，它為Kafka服務器提供生產數據的設施。為了進行演示，讓我們看一下使用docker-compose命令啟動生產者的命令：
$ docker-compose exec kafka kafka-console-producer.sh --topic baeldung
--broker-list kafka:9092在上面的命令中，我們創建了一個 Kafka 生產者來將消息發送到 Kafka 代理。此外，要發送 JSON 數據類型，我們需要調整命令。在繼續之前，我們首先創建一個示例 JSON 文件sampledata.json ：
{
"name": "test",
"age": 26,
"email": "[email protected]",
"city": "Bucharest",
"occupation": "Software Engineer",
"company": "Baeldung Inc.",
"interests": ["programming", "hiking", "reading"]
}上面的sampledata.json文件包含了JSON格式的用戶基本信息。要將 JSON 數據發送到 Kafka 主題，我們需要jq庫，因為它處理 JSON 數據的功能非常強大。為了進行演示，讓我們安裝jq庫以將此 JSON 數據傳遞給 Kafka 生產者：
$ sudo apt-get install jq上面的命令只是在 Linux 機器上安裝jq庫。此外，我們看一下發送 JSON 數據的命令：
$ jq -rc . sampledata.json | docker-compose exec -T kafka kafka-console-producer.sh --topic baeldung --broker-list kafka:9092上面的命令是一個單行命令，用於在 Docker 環境中處理 JSON 數據並將其流式傳輸到 Kafka 主題中。首先， jq命令處理sampledata.json ，然後使用-r選項，確保 JSON 數據為行格式和不帶引號的格式。之後， -c選項確保數據以單行形式呈現，以便數據可以輕鬆地流式傳輸到相應的 Kafka 主題。
5. 消費者數據
至此，我們已經成功將JSON數據發送到baeldung Kafka主題。現在，讓我們看一下使用該數據的命令：
$ docker-compose exec kafka kafka-console-consumer.sh --topic baeldung --from-beginning --bootstrap-server kafka:9092
{"name":"test","age":26,"email":"[email protected]","city":"Bucharest","occupation":"Software Engineer","company":"Baeldung Inc.","interests":["programming","hiking","reading"]}上述命令消耗從一開始發送到baeldung主題的所有數據。在上一節中，我們發送了 JSON 數據。因此，它也使用 JSON 數據。簡而言之，上述命令允許用戶主動監控發送到主題baeldung所有消息。它使用基於 Kafka 的消息傳遞系統促進實時數據消費。
六，結論
在本文中，我們探討瞭如何將 JSON 數據流式傳輸到 Kafka 主題中。首先，我們創建了一個示例 JSON，然後使用生產者將該 JSON 流式傳輸到 Kafka 主題中。之後，我們使用docker-compose命令使用該數據。
簡而言之，我們涵蓋了使用 Kafka 生產者和消費者將 JSON 格式數據發送到主題的所有必要步驟。此外，它還提供模式演化，因為 JSON 可以處理優雅的更新而不影響現有數據。