建立 Cloud Storage Sink 連接器

Cloud Storage 接收器連接器會將資料從 Managed Service for Apache Kafka 串流至 Cloud Storage，讓您以符合成本效益且可擴充的方式，儲存及處理大量資料。

這個連接器的常見用途包括：

  • 資料湖泊擷取。將 Kafka 資料儲存在資料湖泊，以便長期封存及批次處理。

  • 封存資料以符合法規要求。

事前準備

建立 Cloud Storage 接收器連接器前，請確認您具備下列項目：

必要角色和權限

如要取得建立連接器所需的權限，請要求管理員授予您專案的代管 Kafka 連接器編輯者 (roles/managedkafka.connectorEditor) IAM 角色。如要進一步瞭解如何授予角色，請參閱「管理專案、資料夾和組織的存取權」。

這個預先定義的角色具備建立連接器所需的權限。如要查看確切的必要權限，請展開「Required permissions」(必要權限) 部分：

所需權限

您必須具備下列權限，才能建立連接器：

  • 建立連接器： managedkafka.connectors.create

您或許還可透過自訂角色或其他預先定義的角色取得這些權限。

授予 Cloud Storage bucket 寫入權限

代管 Kafka 服務代理必須具備下列權限，才能將訊息寫入 Cloud Storage bucket：

  • storage.objects.create
  • storage.objects.delete

如要授予這些權限，請在包含 Cloud Storage bucket 的專案中，將 Storage 物件使用者 (roles/storage.objectUser) 角色授予服務代理。

如果 Connect 叢集與 Managed Service for Apache Kafka 叢集位於不同專案，請參閱「 在其他專案中建立 Connect 叢集」。

建立 Cloud Storage 接收器連接器

如要建立 Cloud Storage 接收器連接器，請按照下列步驟操作。

控制台

  1. 前往 Google Cloud 控制台的「Connect Clusters」(連結叢集) 頁面。

    前往「Connect Clusters」(連結叢集)

  2. 按一下要建立連接器的 Connect 叢集。

    系統隨即會顯示「Connect cluster details」(連結叢集詳細資料) 頁面。

  3. 按一下「Create connector」(建立連接器)。

    系統會顯示「Create Kafka connector」(建立 Kafka 連接器) 頁面。

  4. 輸入連接器名稱字串。

    如要查看連接器命名準則，請參閱「Managed Service for Apache Kafka 資源命名指南」。

  5. 在「Connector plugin」(連接器外掛程式) 部分，選取「Cloud Storage Sink」。

  6. 指定可串流資料的主題。

  7. 選擇要儲存資料的儲存空間值區。

  8. (選用) 在「Configuration」(設定) 區段中調整其他設定。

  9. 選取「Task restart policy」(任務重新啟動政策)。詳情請參閱「工作重新啟動政策」。

  10. 點選「建立」。

gcloud

執行 gcloud managed-kafka connectors create 指令：

gcloud managed-kafka connectors create CONNECTOR_ID \
    --connect-cluster=CONNECT_CLUSTER_ID \
    --location=LOCATION \
    --config-file=CONFIG_FILE

更改下列內容：

  • CONNECTOR_ID：連接器的 ID 或名稱。
  • CONNECT_CLUSTER_ID：建立連接器 的 Connect 叢集 ID。
  • LOCATION：Connect 叢集的位置。
  • CONFIG_FILE：本機 YAML 或 JSON 設定檔的路徑。

以下是 Cloud Storage 接收器連接器的設定檔範例：

connector.class: "io.aiven.kafka.connect.gcs.GcsSinkConnector"
gcs.bucket.name: "BUCKET_NAME"
gcs.credentials.default: "true"
format.output.type: "json"
key.converter: "org.apache.kafka.connect.storage.StringConverter"
value.converter: "org.apache.kafka.connect.json.JsonConverter"
value.converter.schemas.enable: "false"
topics: "TOPIC_LIST"
tasks.max: "3"

更改下列內容：

  • BUCKET_NAME：要寫入的 Cloud Storage bucket 名稱。

  • TOPIC_LIST：以半形逗號分隔的 Kafka 主題清單，用於指定要讀取的主題。清單中也可以包含規則運算式。

Terraform

您可以使用 Terraform 資源建立連接器。

resource "google_managed_kafka_connector" "example-cloud-storage-sink-connector" {
  project         = data.google_project.default.project_id
  connector_id    = "my-gcs-sink-connector"
  connect_cluster = google_managed_kafka_connect_cluster.default.connect_cluster_id
  location        = "us-central1"

  configs = {
    "connector.class"                = "io.aiven.kafka.connect.gcs.GcsSinkConnector"
    "tasks.max"                      = "3"
    "topics"                         = "GMK_TOPIC_ID"
    "gcs.bucket.name"                = "GCS_BUCKET_NAME"
    "gcs.credentials.default"        = "true"
    "format.output.type"             = "json"
    "name"                           = "my-gcs-sink-connector"
    "value.converter"                = "org.apache.kafka.connect.json.JsonConverter"
    "value.converter.schemas.enable" = "false"
    "key.converter"                  = "org.apache.kafka.connect.storage.StringConverter"
  }
  provider = google-beta
}

如要瞭解如何套用或移除 Terraform 設定，請參閱「基本 Terraform 指令」。

Go

在試用這個範例之前，請先按照「 安裝用戶端程式庫」中的 Go 設定操作說明進行操作。詳情請參閱 Managed Service for Apache Kafka Go API 參考文件。

如要向 Managed Service for Apache Kafka 進行驗證，請設定應用程式預設憑證(ADC)。詳情請參閱「為本機開發環境設定 ADC」。

import (
	"context"
	"fmt"
	"io"

	managedkafka "cloud.google.com/go/managedkafka/apiv1"
	"cloud.google.com/go/managedkafka/apiv1/managedkafkapb"
	"google.golang.org/api/option"
)

// createCloudStorageSinkConnector creates a Cloud Storage Sink connector.
func createCloudStorageSinkConnector(w io.Writer, projectID, region, connectClusterID, connectorID, topics, gcsBucketName, tasksMax, formatOutputType, valueConverter, valueConverterSchemasEnable, keyConverter, gcsCredentialsDefault string, opts ...option.ClientOption) error {
	// TODO(developer): Update with your config values. Here is a sample configuration:
	// projectID := "my-project-id"
	// region := "us-central1"
	// connectClusterID := "my-connect-cluster"
	// connectorID := "GCS_SINK_CONNECTOR_ID"
	// topics := "GMK_TOPIC_ID"
	// gcsBucketName := "GCS_BUCKET_NAME"
	// tasksMax := "3"
	// formatOutputType := "json"
	// valueConverter := "org.apache.kafka.connect.json.JsonConverter"
	// valueConverterSchemasEnable := "false"
	// keyConverter := "org.apache.kafka.connect.storage.StringConverter"
	// gcsCredentialsDefault := "true"
	ctx := context.Background()
	client, err := managedkafka.NewManagedKafkaConnectClient(ctx, opts...)
	if err != nil {
		return fmt.Errorf("managedkafka.NewManagedKafkaConnectClient got err: %w", err)
	}
	defer client.Close()

	parent := fmt.Sprintf("projects/%s/locations/%s/connectClusters/%s", projectID, region, connectClusterID)

	config := map[string]string{
		"connector.class":                "io.aiven.kafka.connect.gcs.GcsSinkConnector",
		"tasks.max":                      tasksMax,
		"topics":                         topics,
		"gcs.bucket.name":                gcsBucketName,
		"gcs.credentials.default":        gcsCredentialsDefault,
		"format.output.type":             formatOutputType,
		"name":                           connectorID,
		"value.converter":                valueConverter,
		"value.converter.schemas.enable": valueConverterSchemasEnable,
		"key.converter":                  keyConverter,
	}

	connector := &managedkafkapb.Connector{
		Name:    fmt.Sprintf("%s/connectors/%s", parent, connectorID),
		Configs: config,
	}

	req := &managedkafkapb.CreateConnectorRequest{
		Parent:      parent,
		ConnectorId: connectorID,
		Connector:   connector,
	}

	resp, err := client.CreateConnector(ctx, req)
	if err != nil {
		return fmt.Errorf("client.CreateConnector got err: %w", err)
	}
	fmt.Fprintf(w, "Created Cloud Storage sink connector: %s\n", resp.Name)
	return nil
}

Java

在試用這個範例之前，請先按照「 安裝用戶端程式庫」中的 Java 設定操作說明進行操作。詳情請參閱 Managed Service for Apache Kafka Java API 參考文件。

如要向 Managed Service for Apache Kafka 進行驗證，請設定應用程式預設憑證。詳情請參閱「 為本機開發環境設定 ADC」。


import com.google.api.gax.rpc.ApiException;
import com.google.cloud.managedkafka.v1.ConnectClusterName;
import com.google.cloud.managedkafka.v1.Connector;
import com.google.cloud.managedkafka.v1.ConnectorName;
import com.google.cloud.managedkafka.v1.CreateConnectorRequest;
import com.google.cloud.managedkafka.v1.ManagedKafkaConnectClient;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

public class CreateCloudStorageSinkConnector {

  public static void main(String[] args) throws Exception {
    // TODO(developer): Replace these variables before running the example.
    String projectId = "my-project-id";
    String region = "my-region"; // e.g. us-east1
    String connectClusterId = "my-connect-cluster";
    String connectorId = "my-gcs-sink-connector";
    String bucketName = "my-gcs-bucket";
    String kafkaTopicName = "kafka-topic";
    String connectorClass = "io.aiven.kafka.connect.gcs.GcsSinkConnector";
    String maxTasks = "3";
    String gcsCredentialsDefault = "true";
    String formatOutputType = "json";
    String valueConverter = "org.apache.kafka.connect.json.JsonConverter";
    String valueSchemasEnable = "false";
    String keyConverter = "org.apache.kafka.connect.storage.StringConverter";
    createCloudStorageSinkConnector(
        projectId,
        region,
        connectClusterId,
        connectorId,
        bucketName,
        kafkaTopicName,
        connectorClass,
        maxTasks,
        gcsCredentialsDefault,
        formatOutputType,
        valueConverter,
        valueSchemasEnable,
        keyConverter);
  }

  public static void createCloudStorageSinkConnector(
      String projectId,
      String region,
      String connectClusterId,
      String connectorId,
      String bucketName,
      String kafkaTopicName,
      String connectorClass,
      String maxTasks,
      String gcsCredentialsDefault,
      String formatOutputType,
      String valueConverter,
      String valueSchemasEnable,
      String keyConverter)
      throws Exception {

    // Build the connector configuration
    Map<String, String> configMap = new HashMap<>();
    configMap.put("connector.class", connectorClass);
    configMap.put("tasks.max", maxTasks);
    configMap.put("topics", kafkaTopicName);
    configMap.put("gcs.bucket.name", bucketName);
    configMap.put("gcs.credentials.default", gcsCredentialsDefault);
    configMap.put("format.output.type", formatOutputType);
    configMap.put("name", connectorId);
    configMap.put("value.converter", valueConverter);
    configMap.put("value.converter.schemas.enable", valueSchemasEnable);
    configMap.put("key.converter", keyConverter);

    Connector connector = Connector.newBuilder()
        .setName(
            ConnectorName.of(projectId, region, connectClusterId, connectorId).toString())
        .putAllConfigs(configMap)
        .build();

    try (ManagedKafkaConnectClient managedKafkaConnectClient = ManagedKafkaConnectClient.create()) {
      CreateConnectorRequest request = CreateConnectorRequest.newBuilder()
          .setParent(ConnectClusterName.of(projectId, region, connectClusterId).toString())
          .setConnectorId(connectorId)
          .setConnector(connector)
          .build();

      // This operation is being handled synchronously.
      Connector response = managedKafkaConnectClient.createConnector(request);
      System.out.printf("Created Cloud Storage Sink connector: %s\n", response.getName());
    } catch (IOException | ApiException e) {
      System.err.printf("managedKafkaConnectClient.createConnector got err: %s\n", e.getMessage());
    }
  }
}

Python

試用這個範例前，請先按照「 安裝用戶端程式庫」中的 Python 設定操作說明進行。詳情請參閱 Managed Service for Apache Kafka Python API 參考文件。

如要向 Managed Service for Apache Kafka 進行驗證，請設定應用程式預設憑證。詳情請參閱「為本機開發環境設定 ADC」。

from google.api_core.exceptions import GoogleAPICallError
from google.cloud.managedkafka_v1.services.managed_kafka_connect import (
    ManagedKafkaConnectClient,
)
from google.cloud.managedkafka_v1.types import Connector, CreateConnectorRequest

connect_client = ManagedKafkaConnectClient()
parent = connect_client.connect_cluster_path(project_id, region, connect_cluster_id)

configs = {
    "connector.class": "io.aiven.kafka.connect.gcs.GcsSinkConnector",
    "tasks.max": tasks_max,
    "topics": topics,
    "gcs.bucket.name": gcs_bucket_name,
    "gcs.credentials.default": "true",
    "format.output.type": format_output_type,
    "name": connector_id,
    "value.converter": value_converter,
    "value.converter.schemas.enable": value_converter_schemas_enable,
    "key.converter": key_converter,
}

connector = Connector()
connector.name = connector_id
connector.configs = configs

request = CreateConnectorRequest(
    parent=parent,
    connector_id=connector_id,
    connector=connector,
)

try:
    operation = connect_client.create_connector(request=request)
    print(f"Waiting for operation {operation.operation.name} to complete...")
    response = operation.result()
    print("Created Connector:", response)
except GoogleAPICallError as e:
    print(f"The operation failed with error: {e}")

建立連接器後，您可以編輯、刪除、暫停、停止或重新啟動連接器。

設定連接器

如需這個連接器專屬屬性的完整清單，請參閱 Cloud Storage 接收器連接器說明文件中的「設定」一節。

輸出欄位

根據預設，連接器只會寫入每筆 Kafka 記錄中的記錄值。如要納入記錄鍵或標頭，請將 format.output.fields 屬性設為以半形逗號分隔的清單，列出要納入的欄位。支援的欄位為 key、value、headers 和 timestamp。例如：

format.output.fields=key,value,headers

輸出格式

連接器支援以 CSV、JSON 和 Parquet 格式寫入 Kafka 記錄。以下各節說明如何設定輸出格式。

CSV 格式

如要將 CSV 檔案寫入 Cloud Storage，請設定 format.output.type=csv。請使用下列轉換器：

  • 金鑰轉換器：org.apache.kafka.connect.storage.StringConverter 或 org.apache.kafka.connect.json.JsonConverter

  • 值轉換器：org.apache.kafka.connect.converters.ByteArrayConverter

JSON 格式

如要將 JSON 格式的檔案寫入 Cloud Storage，請將 format.output.type 設為下列其中一個值：

  • json：每個輸出檔案都包含一個 JSON 物件陣列，每個 JSON 物件代表一筆 Kafka 記錄。

  • jsonl：連接器會將每筆 Kafka 記錄寫入為 JSON 物件，並分行顯示。

以下範例說明連接器如何以 JSON 格式寫入 Kafka 記錄：

{"headers":[{"key":"k1","value":"v1"},{"key":"k2","value":"v2"}],"value":"Hello world","key":"my-key"}

如果將鍵或值轉換器設為 org.apache.kafka.connect.converters.ByteArrayConverter，連接器會將該欄位寫入為 Base64 編碼值。

即使 value.converter.schemas.enable 為 true，連接器也不會在輸出內容中加入 JSON 結構定義。

Parquet 格式

如要將 Parquet 檔案寫入 Cloud Storage，請設定 format.output.type=parquet。

如果鍵或值轉換器為 org.apache.kafka.connect.json.JsonConverter，則 JSON 資料必須包含結構定義。將轉換器的 schemas.enable 屬性設為 true。

如果是 JSON 資料，連接器會將資料寫入 Parquet 檔案，類型為 struct。舉例來說，如果 Kafka 記錄值有名為 userid 的欄位，連接器會寫入名為 value.userid 的資料欄。如要簡化資料欄名稱，請設定 format.output.envelope=false。如果 format.output.envelope 為 false，則 format.output.fields 必須指定單一輸出欄位。

Cloud Storage 檔案設定

下列設定可控管連接器將檔案寫入 Cloud Storage 的方式。

  • gcs.bucket.name：必填，指定連接器寫入資料的 Cloud Storage bucket 名稱。

  • file.compression.type：指定輸出檔案的壓縮類型。例如 gzip、snappy、zstd 和 none。預設值為 none。

  • file.name.prefix：指定要加到每個檔案名稱的前置字串。預設值為空白 ("")。

  • file.name.template：指定檔案名稱的範本。連接器支援範本的預留位置變數，例如主題名稱的 {{topic}} 和分區編號的 {{partition}}。詳情請參閱 Cloud Storage Sink 連接器說明文件。

  • file.max.records：：寫入單一檔案的記錄數量上限。 如果值為 0，則沒有特定上限。預設值為 0。

  • file.max.bytes：要寫入的檔案大小上限 (以位元組為單位)。如果值為 0，則沒有特定上限。預設值為 0。

記錄分組

Cloud Storage 接收器連接器寫入 Kafka 記錄時，預設會依主題和分區將記錄分組。舉例來說，主題 my-topic 的失聯 0 記錄會寫入名為 my-topic-0-OFFSET 的檔案，其中 OFFSET 是起始偏移。

如果 file.name.template 參數包含 {{key}} 預留位置變數，連接器會依索引鍵將記錄分組。在這種模式下，連接器會為每個記錄鍵寫入個別檔案。連接器會使用該鍵的最新記錄覆寫每個檔案。

憑證

gcs.credentials.default 屬性必須設為 true。這項設定可讓連接器自動探索 Google Cloud 憑證。

後續步驟

Apache Kafka® 是 Apache Software Foundation 或其關係企業在美國和/或其他國家/地區的註冊商標。