Kafka Connect 轉換器與錯誤處理機制

– 12 分鐘閱讀

Kafka Connect 提供了強大的轉換器和錯誤處理機制，簡化了資料在 Kafka 與其他系統之間的流動。開發者可以利用內建的轉換器，例如 Filter、Flatten 和 TimestampConverter，對資料進行修改和轉換，滿足不同的資料處理需求。此外，透過設定死信佇列，可以有效地處理錯誤訊息，確保資料管道的穩定性和可靠性。Kafka Connect 的架構由聯結器、任務和工作者組成，它們協同工作，實作了資料的可靠複製和傳輸。位移管理功能可以追蹤已處理的資料，確保資料不會重複處理或遺漏。相較於其他資料整合工具，Kafka Connect 提供了更簡潔易用的方式來建立和管理資料管道，降低了開發和維護的複雜度。

Kafka Connect 深入解析：轉換器與錯誤處理

Kafka Connect 是 Apache Kafka 的一個重要元件，用於實作資料整合與交換。在前面的章節中，我們已經瞭解瞭如何使用 Kafka Connect 來建立資料管道，並介紹了一些基本的組態選項。本章節將探討 Kafka Connect 的轉換器（Transformations）與錯誤處理（Error Handling）機制。

轉換器（Transformations）

轉換器是一種特殊的元件，用於在資料傳輸過程中對資料進行修改或轉換。Kafka Connect 提供了多種內建的轉換器，例如：

  • Filter：根據特定條件過濾訊息。
  • Flatten：將巢狀資料結構扁平化。
  • HeaderFrom：將欄位從訊息複製到標頭。
  • InsertHeader：在訊息標頭中新增靜態字串。
  • InsertField：在訊息中新增新的欄位。
  • RegexRouter：使用正規表示式修改目的地主題。
  • ReplaceField：移除或重新命名訊息中的欄位。
  • TimestampConverter：修改欄位中的時間格式。
  • TimestampRouter：根據訊息時間戳記修改主題。

轉換器的使用範例

以下是一個使用 InsertHeader 轉換器的範例，該轉換器在訊息標頭中新增了一個名為 MessageSource 的欄位，值為 mysql-login-connector：

{
  "name": "mysql-login-connector",
  "config": {
    "connector.class": "JdbcSourceConnector",
    "connection.url": "jdbc:mysql://127.0.0.1:3306/test?user=root",
    "mode": "timestamp",
    "table.whitelist": "login",
    "validate.non.null": "false",
    "timestamp.column.name": "login_time",
    "topic.prefix": "mysql.",
    "transforms": "InsertHeader",
    "transforms.InsertHeader.type": "org.apache.kafka.connect.transforms.InsertHeader",
    "transforms.InsertHeader.header": "MessageSource",
    "transforms.InsertHeader.value.literal": "mysql-login-connector"
  }
}

程式碼解析

{
  "name": "mysql-login-connector",
  "config": {
    "connector.class": "JdbcSourceConnector",
    ......
    "transforms": "InsertHeader",
    "transforms.InsertHeader.type": "org.apache.kafka.connect.transforms.InsertHeader",
    "transforms.InsertHeader.header": "MessageSource",
    "transforms.InsertHeader.value.literal": "mysql-login-connector"
  }
}

程式碼解密：

  1. 設定轉換器名稱："transforms": "InsertHeader" 指定了轉換器的名稱為 InsertHeader，這是使用者自定義的名稱，用於在組態中參照該轉換器。
  2. 指定轉換器類別："transforms.InsertHeader.type": "org.apache.kafka.connect.transforms.InsertHeader" 指定了 InsertHeader 轉換器的具體實作類別，即 Kafka Connect 內建的 InsertHeader 類別，用於在訊息的標頭中插入自定義的鍵值對。
  3. 定義標頭名稱："transforms.InsertHeader.header": "MessageSource" 設定了要插入的標頭鍵名稱為 MessageSource，即在訊息的標頭中新增一個名為 MessageSource 的欄位。
  4. 設定標頭的值："transforms.InsertHeader.value.literal": "mysql-login-connector" 指定了 MessageSource 這個標頭的值為 mysql-login-connector，這是一個靜態值，表示該訊息的來源是 MySQL 登入聯結器。

錯誤處理與死信佇列

Kafka Connect 提供了錯誤處理機制，可以將錯誤的訊息路由到死信佇列（Dead Letter Queue）。死信佇列是一個特殊的主題，用於儲存無法被正確處理的訊息。

死信佇列的組態

要啟用死信佇列，需要在聯結器的組態中新增以下選項：

errors.tolerance = all
errors.deadletterqueue.topic.name = <dead-letter-queue-topic-name>

Kafka Connect 的架構

Kafka Connect 的架構由三個主要元件組成：聯結器（Connectors）、任務（Tasks）和工作者（Workers）。

聯結器（Connectors）

聯結器負責定義資料複製的工作，並將工作分配給任務。聯結器還負責取得任務的組態，並將其傳遞給工作者。

任務（Tasks）

任務負責實際執行資料複製的工作。任務會從外部系統讀取或寫入資料，並將資料傳送到 Kafka 或從 Kafka 接收資料。

工作者（Workers）

工作者是 Kafka Connect 的執行單元，負責執行聯結器和任務。工作者還負責處理 HTTP 請求、管理組態、提供可靠性、高用性和負載平衡。

Kafka Connect：資料整合的強大工具

Kafka Connect 是 Apache Kafka 的一部分，旨在簡化資料在 Kafka 與其他系統之間的流動。它提供了一個統一的框架，用於將資料從各種來源匯入 Kafka，或將 Kafka 中的資料匯出到其他系統。Kafka Connect 的設計目標是使資料整合變得簡單、可靠且可擴充套件。

為何需要 Kafka Connect？

在沒有 Kafka Connect 的情況下，開發人員需要為每個資料來源或目標系統編寫自定義程式碼，以實作資料的匯入和匯出。這種方法不僅耗時耗力，而且容易出錯。舉例來說，建立一個從資料函式庫讀取資料並插入 Kafka 的資料管道可能需要數天或數週的時間，但如果需要處理組態、錯誤處理、REST API、監控、佈署、擴充套件和故障處理等問題，則可能需要數月的時間才能完成。

內容解密：

  • Kafka Connect 簡化了資料整合的過程，減少了開發和維護的複雜度。
  • 透過使用 Kafka Connect，開發人員可以專注於業務邏輯，而無需擔心底層的技術細節。

Converters 和 Connect 的資料模型

Kafka Connect 的資料模型和轉換器（Converters）是其核心組成部分。Connect API 包含一個資料 API，用於描述資料的結構和內容。例如，JDBC 源聯結器讀取資料函式庫中的列，並根據資料型別構建一個 Connect Schema 物件。然後，它使用該 Schema 構建一個 Struct 物件，其中包含資料函式庫記錄中的所有欄位。

// 示例程式碼：使用 JDBC 源聯結器讀取資料函式庫中的資料
// 假設我們使用的是 MySQL 資料函式庫
Properties props = new Properties();
props.put("connection.url", "jdbc:mysql://localhost:3306/mydb");
props.put("connection.user", "myuser");
props.put("connection.password", "mypassword");
props.put("table.whitelist", "mytable");

// 建立 JDBC 源聯結器
JDBCSourceConnector connector = new JDBCSourceConnector();
connector.start(props);

// 讀取資料函式庫中的資料並轉換為 Connect 資料模型
// 這裡省略了具體的實作細節

內容解密：

  • Connect API 使用資料 API 描述資料的結構和內容。
  • JDBC 源聯結器讀取資料函式庫中的列，並根據資料型別構建一個 Connect Schema 物件。
  • Converters 負責將 Connect 資料模型轉換為 Kafka 可以儲存的格式，如 Avro、JSON 或 Protobufs。

位移管理（Offset Management）

Kafka Connect 提供了位移管理功能，以確保資料的可靠處理。源聯結器需要知道已經處理過哪些資料，並使用 Kafka 提供的 API 來維護相關資訊。對於源聯結器，這意味著傳回給 Connect worker 的記錄包含邏輯分割槽和邏輯位移。當 worker 成功將記錄傳送到 Kafka 後，它會儲存這些記錄的位移，以便在重新啟動或當機後可以從最後儲存的位移繼續處理。

# 示例組態：設定位移儲存的主題名稱
offset.storage.topic=connect-offsets

內容解密：

  • 位移管理是 Kafka Connect 的一個重要功能，確保資料的可靠處理。
  • 源聯結器傳回給 worker 的記錄包含邏輯分割槽和邏輯位移。
  • Worker 成功傳送記錄到 Kafka 後，會儲存這些記錄的位移。

Kafka Connect 的替代方案

雖然 Kafka Connect 提供了便利性和可靠性，但它並不是將資料匯入或匯出 Kafka 的唯一方法。其他替代方案包括：

  1. 其他資料儲存系統的 Ingest 框架：如 Hadoop 的 Flume、Elasticsearch 的 Logstash 或 Fluentd。
  2. 根據 GUI 的 ETL 工具：如 Informatica、Talend、Pentaho、Apache NiFi 和 StreamSets。

這些工具在特定場景下可能更適合，例如當 Kafka 不是架構的核心，或者需要與其他系統整合時。

跨叢集資料映象：架構與實務應用

在大多數情況下，Kafka 的設定、維護和使用都集中在單一叢集上。然而，在某些特定的場景下，系統架構可能需要多個 Kafka 叢集。本章將探討跨叢集資料映象的相關議題，涵蓋其使用案例、架構模式、以及實作工具如 MirrorMaker 的介紹與操作技巧。

跨叢集映象的使用案例

跨叢集映象主要應用於以下幾種情境：

  1. 區域性與中央叢集：在不同地理區域或城市設定獨立的 Kafka 叢集，以滿足本地應用需求。同時，將資料映象至中央叢集，以進行跨區域的資料分析和處理。

    • 例如，一家公司在多個城市設有資料中心，各自收集當地的供需資訊，並將這些資料映象至中央叢集，供商業分析師進行全公司的營收報告。

  2. 高用性（HA）與災難復原（DR）：為確保系統的持續運作，在另一個獨立的叢集上複製主要叢集的資料，以備不時之需。

    • 當主要叢集因故無法運作時，可以迅速切換至備用叢集，減少系統停機時間。

映象架構的設計與考量

實施跨叢集映象時，可以採用不同的架構模式，每種模式都有其優缺點：

  • 主動-被動模式：資料僅從主叢集單向同步至備用叢集，適用於災難復原場景。
  • 主動-主動模式：雙向同步資料，適用於需要多地寫入的應用，但需謹慎處理資料衝突問題。

MirrorMaker：Apache Kafka 的跨叢集複製工具

Apache Kafka 提供了 MirrorMaker 作為內建的跨叢集複製工具。以下是使用 MirrorMaker 的基本步驟與操作技巧：

設定 MirrorMaker

  1. 組態來源與目標叢集：指定來源 Kafka 叢集的 topic 以及目標叢集的連線資訊。
  2. 啟動 MirrorMaker：根據組態引數啟動 MirrorMaker 程式，開始資料映象。

效能調校

  • 調整消費者與生產者引數：最佳化從來源叢集讀取資料以及向目標叢集寫入資料的效能。
  • 監控與日誌記錄：定期檢查 MirrorMaker 的運作狀態，並記錄關鍵的日誌資訊，以便故障排除。

替代方案

除了 MirrorMaker 之外，還有其他第三方工具和解決方案可供選擇，例如 Confluent Replicator 等。選擇適合的工具需考量特定的使用場景和效能需求。

結語

跨叢集資料映象是建構高可用性、可擴充套件的 Kafka 系統的重要技術。本章介紹了跨叢集映象的主要使用案例、架構模式，以及如何使用 MirrorMaker 實作跨叢集複製。在實際應用中，需根據具體需求選擇合適的架構和工具，並進行充分的測試和調校，以確保系統的可靠性和效能。

內容解密：

本章節主要講解了 Kafka 在跨叢集資料映象方面的應用。首先介紹了跨叢集映象的使用場景，例如區域性與中央叢集的架構，以及高用性與災難復原的需求。接著，討論了不同的映象架構模式，包括主動-被動和主動-主動模式，並分析了其優缺點。然後，詳細介紹了 Apache Kafka 的內建跨叢集複製工具 MirrorMaker，包括其設定方法、效能調校技巧以及維運注意事項。最後，提到了一些 MirrorMaker 的替代方案，並強調了根據實際需求選擇合適工具的重要性。透過本章節的學習，讀者可以更深入地理解 Kafka 跨叢集資料映象的原理和實務應用，為建構可靠、可擴充套件的 Kafka 系統提供有力的支援。

多叢集架構的應用與實務考量

在現代分散式系統中，Apache Kafka的多叢集架構已成為滿足不同業務需求的重要解決方案。隨著企業擴充套件業務至不同地區或採用多雲策略，多叢集架構能夠提供更高的可用性、彈性及資料治理能力。

多叢集架構的常見應用場景

1. 災難復原與業務連續性

多叢集架構能夠實作跨資料中心的災難復原，確保在單一資料中心發生故障時，系統仍能持續運作。透過在不同地理位置佈署Kafka叢集，可以實作資料的異地備份與快速復原。

2. 法規遵從性

不同國家和地區的法規要求企業採用不同的資料儲存與處理策略。多叢集架構允許企業在不同地區佈署獨立的Kafka叢集，以滿足當地的法規要求，例如資料主權和保留期限等。

3. 雲端遷移與混合雲環境

許多企業同時在本地資料中心和雲端環境中運作業務。多叢集架構能夠實作跨本地資料中心和多個雲端區域的資料同步與共用，提高資料流動效率並降低跨區域傳輸成本。

4. 邊緣叢集的資料聚合

在物聯網（IoT）和零售等產業中，大量資料來自邊緣裝置。多叢集架構透過在中心位置佈署高用性的聚合叢集，能夠有效收集、處理和分析來自眾多邊緣叢集的資料，同時降低對邊緣裝置的連線和儲存要求。

跨資料中心通訊的現實挑戰

在設計多叢集架構時，需要考慮跨資料中心通訊的挑戰：

  1. 高延遲：隨著資料中心之間的物理距離增加，通訊延遲也會上升。
  2. 頻寬限制：廣域網（WAN）的頻寬通常遠低於單一資料中心內的網路頻寬，且可用頻寬可能隨時間波動。
  3. 成本考量：跨資料中心或跨雲傳輸資料通常涉及較高的成本，包括頻寬費用和雲端服務供應商的資料傳輸費用。

多叢集架構的最佳實踐

  1. 每個資料中心至少佈署一個Kafka叢集，避免跨資料中心寫入資料。
  2. 在跨資料中心複製資料時，確保每個事件只被複製一次（除非因錯誤而重試）。
  3. 盡可能從遠端資料中心消費資料，而不是向遠端資料中心生產資料，以減少延遲和網路錯誤的影響。

中心輻射型（Hub-and-Spoke）架構

這是一種常見的多叢集架構模式，適用於多個本地Kafka叢集與一個中央Kafka叢集之間的資料共用場景。其主要優點包括：

  • 資料在本機產生並處理，減少延遲。
  • 事件從各本地叢集單次映象至中央叢集，減少跨區域傳輸的負擔。
  • 支援應用程式在本機處理資料，同時允許某些消費者存取全域資料集。

這種架構特別適合需要在多個地區產生資料、並在中央位置進行匯總處理和分析的場景。

多資料中心架構的設計與挑戰

在現代分散式系統中，多資料中心（Multi-Datacenter）架構是為了提高用性、擴充套件性和災難還原能力而設計的。Kafka作為一個高效的事件流處理平台，支援多資料中心的佈署模式。本文將探討兩種主要的Kafka多資料中心架構：Hub-and-Spoke（中心輻射）架構和Active-Active（雙主）架構。

Hub-and-Spoke 架構

Hub-and-Spoke架構是一種簡單且常見的多資料中心佈署模式。在這種架構中，有一個中心資料中心（Hub），多個區域資料中心（Spoke）將資料複製到中心資料中心。應用程式如果需要處理來自多個資料中心的資料，通常會被佈署在中心資料中心，因為所有事件都會被映象（Mirrored）到該處。

優點

  • 簡單易佈署、組態和監控。
  • 由於每個消費者始終從同一個叢集讀取資料，因此架構簡單。

缺點

  • 區域資料中心的處理器無法存取其他區域資料中心的資料。
  • 對於需要跨區域存取資料的應用場景，這種架構會造成限制。

應用場景

假設某大型銀行在多個城市設有分行，每個城市的Kafka叢集儲存使用者資料和帳戶歷史，並將這些資訊複製到中心叢集進行商業分析。如果使用者存取不同城市的分行，由於使用者資訊不在該城市的叢集中，分行可能無法存取使用者資訊，除非與遠端叢集互動（不推薦）。

Active-Active 架構

Active-Active架構允許多個資料中心分享部分或全部資料，每個資料中心都能生產和消費事件。

優點

  • 允許使用者就近存取資料中心，提高效能和可用性。
  • 提供冗餘和彈性，當一個資料中心不可用時，可以將使用者導向其他可用資料中心。

缺點

  • 在多個位置非同步讀寫資料時，避免衝突是一個挑戰。
  • 需要解決事件映象回饋（Mirroring Feedback）和資料一致性問題。

衝突解決方案

在Active-Active架構中，開發者需要處理非同步更新導致的衝突問題，例如：

  1. 使用者寫入事件到一個資料中心，但讀取事件時從另一個資料中心，可能會看不到最新寫入的事件。
  2. 同一使用者的不同訂單事件可能在不同資料中心被處理，導致衝突。

解決這些衝突的方法包括：

  • 將使用者「固定」到特定的資料中心，除非連線從遠端位置或資料中心不可用。
  • 定義一致的規則來處理衝突事件，例如選擇一個「正確」的事件或處理多個衝突事件。

實作多資料中心架構的考量

在實作Active-Active架構時，需要為每對資料中心設定映象任務，並避免無限迴圈映象事件。解決方案包括為每個邏輯主題在每個資料中心建立獨立的主題，並避免映象來自遠端資料中心的主題。

此圖示說明瞭兩種多資料中心架構的主要特點，包括其優缺點。
玄貓 BlackCat

玄貓 BlackCat

技術愛好者，專注於分享程式開發、雲端技術與 AI 應用的心得體會。