連接器總覽

下表列出 Managed Service for Apache Kafka 支援的 Kafka Connect 連接器類型。您可以使用這些連接器，將 Apache Kafka 與應用程式和其他 Google Cloud服務整合。

轉換者

轉換器負責 Kafka 記錄資料的序列化和反序列化。轉換器會在 Kafka 主題中的原始位元組格式，與 Kafka Connect 使用的內部結構化資料表示法之間進行轉換。

• 如果是接收器連接器，轉換器會將資料從主題的線路格式反序列化為 Kafka Connect 內部資料格式，連接器會使用該格式寫入目標系統。

• 如果是來源連接器，轉換器會將資料從 Kafka Connect 內部資料格式，序列化為 Kafka 主題指定的傳輸格式。

轉換器可確保連接器以與外部系統相容的格式讀取或寫入 Kafka 記錄。

設定連接器時，請設定下列屬性：

• 金鑰轉換器 (key.converter)：序列化及反序列化 Kafka 記錄金鑰時要使用的轉換器。

• 值轉換器 (value.converter)：序列化及反序列化 Kafka 記錄值時使用的轉換器。

如未指定轉換器，預設轉換器類型為 org.apache.kafka.connect.converters.ByteArrayConverter，會以原始位元組格式傳遞資料。

支援的轉換器

Managed Service for Apache Kafka 支援下列內建轉換器：

轉換器的選擇取決於連接器類型和您在 Kafka 中儲存的資料。詳情請參閱特定連接器的說明文件。

Tasks

連接器會建立一或多個平行運作的任務，藉此移轉資料。如要設定連接器建立的工作數量上限，請設定連接器的 tasks.max 設定屬性。連接器建立的工作數量可能會少於這個值。

提高 tasks.max 的值可以提高處理量，但也會增加資源消耗量 (CPU 和記憶體)。最佳值取決於工作負載，以及分配給 Connect 叢集 worker 的資源。如果是接收器連接器，Kafka 主題分區數量也會影響平行處理。

任務重新啟動政策

您可以設定連接器的工作重新啟動政策，決定發生失敗時的行為。連接器支援下列政策：

• 一律不重新啟動。連接器不會重新啟動失敗的工作。這項政策是預設行為。這項功能適用於偵錯，或發生錯誤後需要手動介入的情況。

• 以指數輪詢方式重新啟動。連接器會在延遲一段時間 (稱為「退避」期) 後，重新啟動失敗的工作。每次失敗後，延遲時間會呈指數級增加。建議大多數正式環境工作負載採用這項政策。

  如果使用指數輪詢政策，也請設定最短和最長輪詢時間的值。輪詢時間下限應大於 60 秒，上限應小於 7200 秒。

轉換和述詞

Managed Service for Apache Kafka 支援預設的 Kafka Connect 轉換和述詞。

轉換可讓您修改個別訊息，再傳送至 Managed Service for Apache Kafka (適用於來源連接器) 或外部系統 (適用於接收器連接器)。您可以使用轉換功能遮蓋機密資料、新增時間戳記或重新命名欄位。

述詞可讓您根據特定條件篩選資料，並依據訊息屬性決定轉換作業要套用至哪些訊息。

舉例來說，如要設定接收器連接器，忽略含有 DoNotProcess 標頭鍵的訊息，請新增下列設定：

transforms=dropMessage
transforms.dropMessage.type=org.apache.kafka.connect.transforms.Filter
transforms.dropMessage.predicate=hasKey
predicates=hasKey
predicates.hasKey.type=org.apache.kafka.connect.transforms.predicates.HasHeaderKey
predicates.hasKey.name=DoNotProcess

這項設定會執行下列操作：

1. 設定名為 hasKey 的述詞，類型為 org.apache.kafka.connect.transforms.predicates.HasHeaderKey。這個述詞會比對所有包含鍵 DoNotProcess 標頭的訊息。

2. 設定名為 dropMessage 的轉換，類型為 org.apache.kafka.connect.transforms.Filter。這項轉換會捨棄符合設定述詞的所有訊息。

3. 將轉換連結至述詞 hasKey。這可確保只有含有 DoNotProcess 標頭鍵的訊息會遭到轉換作業捨棄。

後續步驟

• Kafka Connect 總覽

• 建立 Connect 叢集