Google は AI 技術を使用して、コンテンツをご希望の言語に翻訳しています。AI 翻訳には誤りが含まれる場合があります。

Cloud Storage Sink コネクタを作成する

Cloud Storage シンクコネクタは、Managed Service for Apache Kafka から Cloud Storage にデータをストリーミングします。これにより、大量のデータを費用対効果の高いスケーラブルな方法で保存して処理できます。

このコネクタの一般的なユースケースは次のとおりです。

データレイクの取り込み。長期アーカイブとバッチ処理のために Kafka データをデータレイクに保存します。
規制要件を満たすためにデータをアーカイブする。

始める前に

Cloud Storage Sink コネクタを作成する前に、次のものが揃っていることを確認してください。

Connect クラスタ用の Managed Service for Apache Kafka クラスタを作成します。これは、Connect クラスタに関連付けられているプライマリ Kafka クラスタです。これは、コネクタパイプラインの一端を形成するソースクラスタでもあります。
Cloud Storage シンクコネクタをホストする Connect クラスタを作成します。
Kafka からストリーミングされたデータを保存する Cloud Storage バケットを作成します。
ソースクラスタ内に Kafka トピックを作成して構成します。データは、この Kafka トピックから宛先の Cloud Storage バケットに移動します。

必要なロールと権限

コネクタの作成に必要な権限を取得するには、プロジェクトに対するマネージド Kafka コネクタ編集者（roles/managedkafka.connectorEditor）IAM ロールを付与するよう管理者に依頼してください。ロールの付与については、プロジェクト、フォルダ、組織に対するアクセス権の管理をご覧ください。

この事前定義ロールには、コネクタの作成に必要な権限が含まれています。必要とされる正確な権限については、「必要な権限」セクションを開いてご確認ください。

必要な権限

コネクタを作成するには、次の権限が必要です。

コネクタを作成します。 managedkafka.connectors.create

カスタムロールや他の事前定義ロールを使用して、これらの権限を取得することもできます。

Cloud Storage バケットへの書き込み権限を付与する

マネージド Kafka サービスエージェントには、Cloud Storage バケットにメッセージを書き込むための次の権限が必要です。

storage.objects.create
storage.objects.delete

これらの権限を付与するには、Cloud Storage バケットを含むプロジェクトに対する Storage オブジェクトユーザー（roles/storage.objectUser）ロールをサービスエージェントに付与します。

Connect クラスタが Managed Service for Apache Kafka クラスタとは異なるプロジェクトにある場合は、別のプロジェクトに Connect クラスタを作成するをご覧ください。

Cloud Storage シンクコネクタを作成する

Cloud Storage シンクコネクタを作成する手順は次のとおりです。

コンソール

Google Cloud コンソールで、[Connect クラスタ] ページに移動します。

[Connect クラスタ] に移動
コネクタを作成する Connect クラスタをクリックします。

[Connect クラスタの詳細] ページが表示されます。
[コネクタを作成] をクリックします。

[Kafka コネクタの作成] ページが表示されます。
コネクタ名に文字列を入力します。

コネクタの命名方法のガイドラインについては、Managed Service for Apache Kafka リソースの命名ガイドラインをご覧ください。
[コネクタプラグイン] で [Cloud Storage シンク] を選択します。
データをストリーミングできるトピックを指定します。
データを保存するストレージバケットを選択します。
（省略可）[構成] セクションで追加の設定を行います。
[タスクの再起動ポリシー] を選択します。詳細については、タスクの再起動ポリシーをご覧ください。
[作成] をクリックします。

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 Sink コネクタの構成ファイルの例を次に示します。

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 バケットの名前。
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 リファレンスドキュメントをご覧ください。

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 シンクコネクタのドキュメントの構成をご覧ください。

Output fields

デフォルトでは、コネクタは各 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 オブジェクトの配列が 1 つ含まれています。各 JSON オブジェクトは 1 つの Kafka レコードを表します。
jsonl: コネクタは、各 Kafka レコードを個別の行に JSON オブジェクトとして書き込みます。

次の例は、コネクタが Kafka レコードを JSON 形式で書き込む方法を示しています。

{"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 バケットの名前を指定します。
file.compression.type: 出力ファイルの圧縮タイプを指定します。例: gzip、snappy、zstd、none。デフォルト値は none です。
file.name.prefix: 各ファイルの名前に追加する接頭辞を指定します。デフォルト値は空（""）です。
file.name.template: ファイル名のテンプレートを指定します。このコネクタは、テンプレートのプレースホルダ変数（トピック名の {{topic}} やパーティション番号の {{partition}} など）をサポートしています。詳細については、Cloud Storage Sink コネクタのドキュメントをご覧ください。
file.max.records: 1 つのファイルに書き込むレコードの最大数。値が 0 の場合、特定の最大値はありません。デフォルト値は 0 です。
file.max.bytes: 書き込むファイルの最大サイズ（バイト単位）。値が 0 の場合、特定の最大値はありません。デフォルト値は 0 です。

レコードのグループ化

Cloud Storage Sink コネクタが Kafka レコードを書き込む場合、デフォルトではレコードをトピックとパーティションでグループ化します。たとえば、トピック my-topic のパーティション 0 のレコードは、my-topic-0-OFFSET という名前のファイルに書き込まれます。ここで、OFFSET は開始オフセットです。

file.name.template パラメータに {{key}} プレースホルダ変数が含まれている場合、コネクタはレコードをキーでグループ化します。このモードでは、コネクタはレコードキーごとに個別のファイルを書き込みます。コネクタは、各ファイルをそのキーの最新のレコードで上書きします。

認証情報

gcs.credentials.default プロパティは true に設定する必要があります。この設定により、コネクタは Google Cloud 認証情報を自動的に検出できます。

次のステップ

Apache Kafka® は、Apache Software Foundation または米国その他の諸国における関連会社の商標です。

Cloud Storage Sink コネクタを作成する コレクションでコンテンツを整理 必要に応じて、コンテンツの保存と分類を行います。