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

Pub/Sub Sink コネクタを作成する

Pub/Sub Sink コネクタは、Kafka トピックから Pub/Sub トピックにメッセージをストリーミングします。これにより、Kafka ベースのアプリケーションを Pub/Sub と統合できます。

Pub/Sub シンクコネクタのユースケースには、次のようなものがあります。

イベントドリブンアーキテクチャ。Kafka にパブリッシュされたメッセージから Pub/Sub 処理をトリガーする。
サービスの統合。Kafka から Pub/Sub を使用する他の Google Cloud サービスまたはアプリケーションにデータを送信する。
処理されたデータに基づいて、リアルタイムの通知やアクションをトリガーします。

始める前に

Pub/Sub シンクコネクタを作成する前に、次のものが揃っていることを確認してください。

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

必要なロールと権限

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

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

必要な権限

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

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

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

Pub/Sub トピックにパブリッシュする権限を付与する

マネージド Kafka サービスエージェントには、Pub/Sub トピックにメッセージをパブリッシュする権限が必要です。Pub/Sub トピックを含むプロジェクトに対する Pub/Sub パブリッシャー（roles/pubsub.publisher）ロールをサービスエージェントに付与します。

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

Pub/Sub シンクコネクタの仕組み

Pub/Sub シンクコネクタは、1 つ以上の Kafka トピックからメッセージを pull し、Pub/Sub トピックにパブリッシュします。

Pub/Sub シンクコネクタがデータをコピーする方法の詳細は次のとおりです。

コネクタは、ソースクラスタ内の 1 つ以上の Kafka トピックからメッセージを消費します。
コネクタは、cps.topic 構成プロパティを使用して指定されたターゲット Pub/Sub トピック ID にメッセージを書き込みます。これは必須プロパティです。
また、コネクタでは、cps.project 構成プロパティを使用して、Pub/Sub トピックを含む Google Cloud プロジェクトを指定する必要があります。これは必須プロパティです。
コネクタは、cps.endpoint プロパティを使用して指定されたカスタム Pub/Sub エンドポイントを使用することもできます。デフォルトのエンドポイントは "pubsub.googleapis.com:443" です。
パフォーマンスを最適化するため、コネクタは Pub/Sub にパブリッシュする前にメッセージをバッファリングします。maxBufferSize、maxBufferBytes、maxDelayThresholdMs、maxOutstandingRequestBytes、maxOutstandingMessages を構成して、バッファリングを制御できます。
Kafka レコードには、ヘッダー、キー、値の 3 つのコンポーネントがあります。コネクタは、キーと値のコンバータを使用して、Kafka メッセージデータを Pub/Sub で想定される形式に変換します。構造体またはマップ値スキーマを使用する場合、messageBodyName プロパティは、Pub/Sub メッセージ本文として使用するフィールドまたはキーを指定します。
コネクタは、metadata.publish プロパティを true に設定することで、Kafka トピック、パーティション、オフセット、タイムスタンプをメッセージ属性として含めることができます。
コネクタは、headers.publish プロパティを true に設定することで、Kafka メッセージヘッダーを Pub/Sub メッセージ属性として含めることができます。
コネクタには、orderingKeySource プロパティを使用して Pub/Sub メッセージの順序指定キーを含めることができます。値のオプションには、"none"（デフォルト）、"key"、"partition" があります。
tasks.max プロパティは、コネクタの並列処理のレベルを制御します。tasks.max を増やすとスループットが向上しますが、実際の並列処理は Kafka トピックのパーティション数によって制限されます。

Pub/Sub シンクコネクタのプロパティ

Pub/Sub シンクコネクタを作成するときに、次のプロパティを指定する必要があります。

コネクタ名

Connect クラスタ内で一意のコネクタの名前。リソースの命名に関するガイドラインについては、Managed Service for Apache Kafka リソースの命名ガイドラインをご覧ください。

コネクタプラグインの種類

コネクタプラグインタイプとして [Pub/Sub Sink] を選択します。これにより、Kafka から Pub/Sub へのデータフローの方向と、使用される特定のコネクタ実装が決定されます。ユーザーインターフェースを使用してコネクタを構成しない場合は、コネクタクラスも指定する必要があります。

Kafka トピック

コネクタがメッセージを消費する Kafka トピック。1 つ以上のトピックを指定することも、正規表現を使用して複数のトピックを照合することもできます。たとえば、topic.* を使用すると、「topic」で始まるすべてのトピックを照合できます。これらのトピックは、Connect クラスタに関連付けられた Managed Service for Apache Kafka クラスタ内に存在する必要があります。

Pub/Sub トピック

コネクタがメッセージをパブリッシュする既存の Pub/Sub トピック。始める前にで説明されているように、Connect クラスタのサービスアカウントに、トピックのプロジェクトに対する roles/pubsub.publisher ロールがあることを確認します。

構成

このセクションでは、コネクタ固有の追加の構成プロパティを指定できます。

Kafka トピックのデータは Avro、JSON、未加工バイトなどのさまざまな形式にできるため、構成の重要な部分にはコンバータの指定が含まれます。コンバータは、Kafka トピックで使用されている形式から Kafka Connect の標準化された内部形式にデータを変換します。Pub/Sub シンクコネクタは、この内部データを取得し、Pub/Sub で必要な形式に変換してから書き込みます。

Kafka Connect でのコンバータの役割、サポートされているコンバータの種類、一般的な構成オプションの詳細については、コンバータをご覧ください。

Pub/Sub シンクコネクタに固有の構成を次に示します。

cps.project: Pub/Sub トピックを含む Google Cloud プロジェクト ID を指定します。
cps.topic: データが公開される Pub/Sub トピックを指定します。
cps.endpoint: 使用する Pub/Sub エンドポイントを指定します。

このコネクタに固有の使用可能な構成プロパティの一覧については、Pub/Sub シンクコネクタの構成をご覧ください。

Pub/Sub シンクコネクタを作成する

コネクタを作成する前に、Pub/Sub シンクコネクタのプロパティのドキュメントを確認してください。

コンソール

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

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

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

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

コネクタの命名方法のガイドラインについては、Managed Service for Apache Kafka リソースの命名ガイドラインをご覧ください。
[コネクタプラグイン] で、[Pub/Sub シンク] を選択します。
[トピック] で、[Kafka トピックのリストを選択する] または [トピックの正規表現を使用する] を選択します。次に、このコネクタがメッセージを消費する Kafka トピックを選択または入力します。これらのトピックは、関連付けられた Kafka クラスタにあります。
[Cloud Pub/Sub トピックを選択してください] で、このコネクタがメッセージをパブリッシュする Pub/Sub トピックを選択します。トピックは、完全なリソース名の形式（projects/{project}/topics/{topic}）で表示されます。
（省略可）[構成] セクションで追加の設定を行います。前のセクションで説明したように、tasks.max、key.converter、value.converter などのプロパティを指定します。
[タスクの再起動ポリシー] を選択します。詳細については、タスクの再起動ポリシーをご覧ください。
[作成] をクリックします。

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 構成ファイルのパス。

Pub/Sub シンクコネクタの構成ファイルの例を次に示します。

connector.class: "com.google.pubsub.kafka.sink.CloudPubSubSinkConnector"
name: "CPS_SINK_CONNECTOR_ID"
tasks.max: "1"
topics: "GMK_TOPIC_ID"
value.converter: "org.apache.kafka.connect.storage.StringConverter"
key.converter: "org.apache.kafka.connect.storage.StringConverter"
cps.topic: "CPS_TOPIC_ID"
cps.project: "GCP_PROJECT_ID"

次のように置き換えます。

CPS_SINK_CONNECTOR_ID: Pub/Sub Sink コネクタの ID または名前。コネクタの命名方法のガイドラインについては、Managed Service for Apache Kafka リソースの命名ガイドラインをご覧ください。コネクタの名前は変更できません。
GMK_TOPIC_ID: Pub/Sub Sink コネクタがデータを読み取る Managed Service for Apache Kafka トピックの ID。
CPS_TOPIC_ID: データが公開される Pub/Sub トピックの ID。
GCP_PROJECT_ID: Pub/Sub トピックが存在する Google Cloudプロジェクトの ID。

Terraform

コネクタの作成には Terraform リソースを使用します。

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

  configs = {
    "connector.class" = "com.google.pubsub.kafka.sink.CloudPubSubSinkConnector"
    "name"            = "my-pubsub-sink-connector"
    "tasks.max"       = "3"
    "topics"          = "TOPIC_NAME"
    "cps.topic"       = "CPS_TOPIC_NAME"
    "cps.project"     = "CPS_PROJECT_NAME"
    "value.converter" = "org.apache.kafka.connect.storage.StringConverter"
    "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"
)

// createPubSubSinkConnector creates a Pub/Sub Sink connector.
func createPubSubSinkConnector(w io.Writer, projectID, region, connectClusterID, connectorID, topics, valueConverter, keyConverter, cpsTopic, cpsProject, tasksMax 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 := "CPS_SINK_CONNECTOR_ID"
	// topics := "GMK_TOPIC_ID"
	// valueConverter := "org.apache.kafka.connect.storage.StringConverter"
	// keyConverter := "org.apache.kafka.connect.storage.StringConverter"
	// cpsTopic := "CPS_TOPIC_ID"
	// cpsProject := "GCP_PROJECT_ID"
	// tasksMax := "3"
	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)

	// Pub/Sub Sink sample connector configuration
	config := map[string]string{
		"connector.class": "com.google.pubsub.kafka.sink.CloudPubSubSinkConnector",
		"name":            connectorID,
		"tasks.max":       tasksMax,
		"topics":          topics,
		"value.converter": valueConverter,
		"key.converter":   keyConverter,
		"cps.topic":       cpsTopic,
		"cps.project":     cpsProject,
	}

	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 Pub/Sub 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 CreatePubSubSinkConnector {

  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-pubsub-sink-connector";
    String pubsubProjectId = "my-pubsub-project-id";
    String pubsubTopicName = "my-pubsub-topic";
    String kafkaTopicName = "kafka-topic";
    String connectorClass = "com.google.pubsub.kafka.sink.CloudPubSubSinkConnector";
    String maxTasks = "3";
    String valueConverter = "org.apache.kafka.connect.storage.StringConverter";
    String keyConverter = "org.apache.kafka.connect.storage.StringConverter";
    createPubSubSinkConnector(
        projectId,
        region,
        connectClusterId,
        connectorId,
        pubsubProjectId,
        pubsubTopicName,
        kafkaTopicName,
        connectorClass,
        maxTasks,
        valueConverter,
        keyConverter);
  }

  public static void createPubSubSinkConnector(
      String projectId,
      String region,
      String connectClusterId,
      String connectorId,
      String pubsubProjectId,
      String pubsubTopicName,
      String kafkaTopicName,
      String connectorClass,
      String maxTasks,
      String valueConverter,
      String keyConverter)
      throws Exception {

    // Build the connector configuration
    Map<String, String> configMap = new HashMap<>();
    configMap.put("connector.class", connectorClass);
    configMap.put("name", connectorId);
    configMap.put("tasks.max", maxTasks);
    configMap.put("topics", kafkaTopicName);
    configMap.put("value.converter", valueConverter);
    configMap.put("key.converter", keyConverter);
    configMap.put("cps.topic", pubsubTopicName);
    configMap.put("cps.project", pubsubProjectId);

    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 Pub/Sub 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": "com.google.pubsub.kafka.sink.CloudPubSubSinkConnector",
    "name": connector_id,
    "tasks.max": tasks_max,
    "topics": topics,
    "value.converter": value_converter,
    "key.converter": key_converter,
    "cps.topic": cps_topic,
    "cps.project": cps_project,
}

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}")

コネクタを作成したら、コネクタの編集、削除、一時停止、停止、再起動を行うことができます。

次のステップ

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

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