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

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

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

Cloud Storage シンクコネクタのユースケースは次のとおりです。

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

始める前に

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

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

必要なロールと権限

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

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

必要な権限

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

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

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

Cloud Storage バケットに書き込む権限を付与する

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

storage.objects.create
storage.objects.delete

Cloud Storage バケットを含むプロジェクトのサービスアカウントに Storage オブジェクトユーザー（roles/storage.objectUser）ロールを付与します。

Managed Kafka サービスアカウントの形式は service-PROJECT_NUMBER@gcp-sa-managedkafka.iam.gserviceaccount.com です。ここで、PROJECT_NUMBER は Connect クラスタのプロジェクト番号です。

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

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

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

コンソール

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

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

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

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

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

gcloud

Google Cloud コンソールで Cloud Shell をアクティブにします。

Cloud Shell をアクティブにする

Google Cloud コンソールの下部で Cloud Shell セッションが開始し、コマンドラインプロンプトが表示されます。Cloud Shell はシェル環境です。Google Cloud CLI がすでにインストールされており、現在のプロジェクトの値もすでに設定されています。セッションが初期化されるまで数秒かかることがあります。
gcloud managed-kafka connectors create コマンドを実行します。
```
gcloud managed-kafka connectors create CONNECTOR_ID \
    --location=LOCATION \
    --connect-cluster=CONNECT_CLUSTER_ID \
    --config-file=CONFIG_FILE
```
次のように置き換えます。
- CONNECTOR_ID: コネクタの ID または名前。コネクタの命名方法のガイドラインについては、Managed Service for Apache Kafka リソースの命名ガイドラインをご覧ください。コネクタの名前は変更できません。
- LOCATION: コネクタを作成するロケーション。これは、Connect クラスタを作成したロケーションと同じである必要があります。
- CONNECT_CLUSTER_ID: コネクタが作成される Connect クラスタの ID。
- CONFIG_FILE: コネクタの YAML 構成ファイルへのパス。
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 リファレンスドキュメントをご覧ください。

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

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 オブジェクトの単一の配列が含まれています。各 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 コネクタのドキュメントをご覧ください。

レコードのグループ化

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 コネクタを作成する コレクションでコンテンツを整理 必要に応じて、コンテンツの保存と分類を行います。