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

BigQuery Sink コネクタを作成する

BigQuery Sink コネクタを使用すると、Managed Service for Apache Kafka から BigQuery テーブルにデータをリアルタイムでストリーミングできます。このコネクタを使用すると、ストリーミングデータを BigQuery に取り込み、データウェアハウジング、分析レポート、リアルタイムダッシュボードの強化を行うことができます。

始める前に

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

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

必要なロールと権限

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

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

必要な権限

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

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

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

BigQuery テーブルへの書き込み権限を付与する

マネージド Kafka サービスエージェントには、BigQuery テーブルにメッセージを書き込む権限が必要です。テーブルを含むプロジェクトに対する BigQuery データ編集者（roles/bigquery.dataEditor）のロールをサービスエージェントに付与します。

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

主な構成パラメータ

BigQuery シンクコネクタの構成に関する推奨事項は次のとおりです。

キューサイズ

queueSize 設定は、BigQuery への書き込みリクエストの最大数を制御します。デフォルトでは、この値は構成されていません。これにより、Connect クラスタでメモリ不足エラーが発生する可能性があります。このようなエラーの発生を減らすため、この値を構成することをおすすめします。キュー内のアイテムは書き込みリクエストであり、メッセージ数やバイト数に直接対応するものではありません。そのため、推奨値を事前に把握することは困難です。

Kafka からメッセージがポーリングされると、コネクタは宛先テーブルに基づいて作業を分割します。ポーリング内のすべてのメッセージを 1 つのテーブルに書き込む場合は、1 つの作業項目のみが追加されます。ただし、ポーリング内のメッセージが 3 つのテーブルに書き込まれる場合は、3 つの作業項目が追加されます。ワークアイテムのサイズは、メッセージポーリング構成、Kafka メッセージのサイズ、書き込み先のテーブル数に基づいて大まかに見積もることができます。

BigQuery シンクコネクタのスキーマ

BigQuery Sink コネクタは、構成された値コンバータ（value.converter）を使用して、Kafka レコード値をフィールドに解析します。次に、BigQuery テーブルの同じ名前の列にフィールドを書き込みます。

コネクタを動作させるにはスキーマが必要です。スキーマは次の方法で指定できます。

メッセージベースのスキーマ: スキーマは各メッセージの一部として含まれます。
テーブルベースのスキーマ: コネクタは、BigQuery テーブルスキーマからメッセージスキーマを推測します。
スキーマレジストリ: コネクタは、Managed Service for Apache Kafka スキーマレジストリ（プレビュー）などのスキーマレジストリからスキーマを読み取ります。

以降のセクションでは、これらのオプションについて説明します。

メッセージベースのスキーマ

このモードでは、各 Kafka レコードに JSON スキーマが含まれます。コネクタは、スキーマを使用してレコードデータを BigQuery テーブル行として書き込みます。

メッセージベースのスキーマを使用するには、コネクタで次のプロパティを設定します。

value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=true

Kafka レコード値の例:

{
  "schema": {
    "type": "struct",
    "fields": [
      {
        "field": "user",
        "type": "string",
        "optional": false
      },
      {
        "field": "age",
        "type": "int64",
        "optional": false
      }
    ]
  },
  "payload": {
    "user": "userId",
    "age": 30
  }
}

宛先テーブルがすでに存在する場合、BigQuery テーブルのスキーマは、埋め込みメッセージスキーマと互換性がある必要があります。autoCreateTables=true の場合、コネクタは必要に応じて宛先テーブルを自動的に作成します。詳細については、テーブルの作成をご覧ください。

メッセージスキーマの変更に応じてコネクタが BigQuery テーブルスキーマを更新するようにするには、allowNewBigQueryFields、allowSchemaUnionization、または allowBigQueryRequiredFieldRelaxation を true に設定します。

テーブルベースのスキーマ

このモードでは、Kafka レコードに明示的なスキーマのないプレーンな JSON データが含まれます。コネクタは、宛先テーブルからスキーマを推測します。

要件:

BigQuery テーブルがすでに存在している必要があります。
Kafka レコードデータはテーブルスキーマと互換性がある必要があります。
このモードでは、着信したメッセージに基づく動的スキーマ更新はサポートされていません。

テーブルベースのスキーマを使用するには、コネクタで次のプロパティを設定します。

value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false
bigQueryPartitionDecorator=false

BigQuery テーブルで日単位のパーティショニングで時間ベースのパーティショニングを使用している場合、bigQueryPartitionDecorator は true にできます。それ以外の場合は、このプロパティを false に設定します。

Kafka レコード値の例:

{
  "user": "userId",
  "age": 30
}

スキーマレジストリ

このモードでは、各 Kafka レコードに Apache Avro データが含まれ、メッセージスキーマはスキーマレジストリに保存されます。

スキーマレジストリで BigQuery シンクコネクタを使用するには、コネクタに次のプロパティを設定します。

value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=SCHEMA_REGISTRY_URL

SCHEMA_REGISTRY_URL は、スキーマレジストリの URL に置き換えます。

Managed Service for Apache Kafka スキーマレジストリでコネクタを使用するには、次のプロパティを設定します。

value.converter.bearer.auth.credentials.source=GCP

詳細については、スキーマレジストリで Kafka Connect を使用するをご覧ください。

Apache Iceberg マネージドテーブル

BigQuery Sink コネクタは、シンクターゲットとして Apache Iceberg マネージドテーブル（以下、Iceberg マネージドテーブル）をサポートしています。

Iceberg マネージドテーブルは、 Google Cloudでオープン形式のレイクハウスを構築するための基盤になります。Iceberg マネージドテーブルは BigQuery テーブルと同じフルマネージドエクスペリエンスを提供します。また、Parquet を使用してお客様所有のストレージバケットにデータを保存することで、オープン形式の Apache Iceberg テーブル形式と相互運用が可能です。

Apache Iceberg テーブルの作成方法については、Apache Iceberg テーブルを作成するをご覧ください。

BigQuery シンクコネクタを作成する

コンソール

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

[Connect クラスタ] に移動
コネクタを作成する Connect クラスタをクリックします。
[コネクタを作成] をクリックします。
コネクタ名に文字列を入力します。

コネクタの命名方法のガイドラインについては、Managed Service for Apache Kafka リソースの命名ガイドラインをご覧ください。
[コネクタプラグイン] で、[BigQuery シンク] を選択します。
[トピック] セクションで、読み取る Kafka トピックを指定します。トピック名のリストまたはトピック名と照合する正規表現を指定できます。
- オプション 1: [Kafka トピックのリストを選択する] を選択します。[Kafka トピック] リストで、1 つ以上のトピックを選択します。[OK] をクリックします。
- オプション 2: [トピックの正規表現を使用する] を選択します。[トピックの正規表現] フィールドに正規表現を入力します。
[データセット] をクリックし、BigQuery データセットを指定します。既存のデータセットを選択することも、新しいデータセットを作成することもできます。
省略可: [構成] ボックスで、構成プロパティを追加するか、デフォルトのプロパティを編集します。詳細については、コネクタを構成するをご覧ください。
[タスクの再起動ポリシー] を選択します。詳細については、タスクの再起動ポリシーをご覧ください。
[作成] をクリックします。

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: BigQuery Sink コネクタの YAML 構成ファイルのパス。
BigQuery Sink コネクタの構成ファイルの例を次に示します。
```
name: "BQ_SINK_CONNECTOR_ID"
project: "GCP_PROJECT_ID"
topics: "GMK_TOPIC_ID"
tasks.max: 3
connector.class: "com.wepay.kafka.connect.bigquery.BigQuerySinkConnector"
key.converter: "org.apache.kafka.connect.storage.StringConverter"
value.converter: "org.apache.kafka.connect.json.JsonConverter"
value.converter.schemas.enable: "false"
defaultDataset: "BQ_DATASET_ID"
```
次のように置き換えます。
- BQ_SINK_CONNECTOR_ID: BigQuery Sink コネクタの ID または名前。コネクタの命名方法のガイドラインについては、Managed Service for Apache Kafka リソースの命名ガイドラインをご覧ください。コネクタの名前は変更できません。
- GCP_PROJECT_ID: BigQuery データセットが存在する Google Cloudプロジェクトの ID。
- GMK_TOPIC_ID: データが BigQuery シンクコネクタに流れる Managed Service for Apache Kafka トピックの ID。
- BQ_DATASET_ID: パイプラインのシンクとして機能する BigQuery データセットの ID。

Terraform

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

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

  configs = {
    "name"                           = "my-bigquery-sink-connector"
    "project"                        = data.google_project.default.project_id
    "topics"                         = "GMK_TOPIC_ID"
    "tasks.max"                      = "3"
    "connector.class"                = "com.wepay.kafka.connect.bigquery.BigQuerySinkConnector"
    "key.converter"                  = "org.apache.kafka.connect.storage.StringConverter"
    "value.converter"                = "org.apache.kafka.connect.json.JsonConverter"
    "value.converter.schemas.enable" = "false"
    "defaultDataset"                 = "BQ_DATASET_ID"
  }

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

// createBigQuerySinkConnector creates a BigQuery Sink connector.
func createBigQuerySinkConnector(w io.Writer, projectID, region, connectClusterID, connectorID, topics, tasksMax, keyConverter, valueConverter, valueConverterSchemasEnable, defaultDataset 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 := "BQ_SINK_CONNECTOR_ID"
	// topics := "GMK_TOPIC_ID"
	// tasksMax := "3"
	// keyConverter := "org.apache.kafka.connect.storage.StringConverter"
	// valueConverter := "org.apache.kafka.connect.json.JsonConverter"
	// valueConverterSchemasEnable := "false"
	// defaultDataset := "BQ_DATASET_ID"
	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)

	// BigQuery Sink sample connector configuration
	config := map[string]string{
		"name":                           connectorID,
		"project":                        projectID,
		"topics":                         topics,
		"tasks.max":                      tasksMax,
		"connector.class":                "com.wepay.kafka.connect.bigquery.BigQuerySinkConnector",
		"key.converter":                  keyConverter,
		"value.converter":                valueConverter,
		"value.converter.schemas.enable": valueConverterSchemasEnable,
		"defaultDataset":                 defaultDataset,
	}

	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 BigQuery 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 CreateBigQuerySinkConnector {

  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-bigquery-sink-connector";
    String bigqueryProjectId = "my-bigquery-project-id";
    String datasetName = "my-dataset";
    String kafkaTopicName = "kafka-topic";
    String maxTasks = "3";
    String connectorClass = "com.wepay.kafka.connect.bigquery.BigQuerySinkConnector";
    String keyConverter = "org.apache.kafka.connect.storage.StringConverter";
    String valueConverter = "org.apache.kafka.connect.json.JsonConverter";
    String valueSchemasEnable = "false";
    createBigQuerySinkConnector(
        projectId,
        region,
        connectClusterId,
        connectorId,
        bigqueryProjectId,
        datasetName,
        kafkaTopicName,
        maxTasks,
        connectorClass,
        keyConverter,
        valueConverter,
        valueSchemasEnable);
  }

  public static void createBigQuerySinkConnector(
      String projectId,
      String region,
      String connectClusterId,
      String connectorId,
      String bigqueryProjectId,
      String datasetName,
      String kafkaTopicName,
      String maxTasks,
      String connectorClass,
      String keyConverter,
      String valueConverter,
      String valueSchemasEnable)
      throws Exception {

    // Build the connector configuration
    Map<String, String> configMap = new HashMap<>();
    configMap.put("name", connectorId);
    configMap.put("project", bigqueryProjectId);
    configMap.put("topics", kafkaTopicName);
    configMap.put("tasks.max", maxTasks);
    configMap.put("connector.class", connectorClass);
    configMap.put("key.converter", keyConverter);
    configMap.put("value.converter", valueConverter);
    configMap.put("value.converter.schemas.enable", valueSchemasEnable);
    configMap.put("defaultDataset", datasetName);

    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 BigQuery 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 = {
    "name": connector_id,
    "project": project_id,
    "topics": topics,
    "tasks.max": tasks_max,
    "connector.class": "com.wepay.kafka.connect.bigquery.BigQuerySinkConnector",
    "key.converter": key_converter,
    "value.converter": value_converter,
    "value.converter.schemas.enable": value_converter_schemas_enable,
    "defaultDataset": default_dataset,
}

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

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

コネクタを構成する

このセクションでは、コネクタに設定できる構成プロパティについて説明します。このコネクタに固有のプロパティの一覧については、BigQuery Sink コネクタの構成をご覧ください。

テーブル名

デフォルトでは、コネクタはトピック名を BigQuery テーブル名として使用します。別のテーブル名を使用するには、次の形式で topic2TableMap プロパティを設定します。

topic2TableMap=TOPIC_1:TABLE_1,TOPIC_2:TABLE_2,...

テーブルの作成

BigQuery シンクコネクタは、宛先テーブルが存在しない場合に作成できます。

autoCreateTables=true の場合、コネクタは存在しない BigQuery テーブルの作成を試みます。この設定がデフォルトの動作です。
autoCreateTables=false の場合、コネクタはテーブルを作成しません。宛先テーブルが存在しない場合は、エラーが発生します。

autoCreateTables が true の場合、次の構成プロパティを使用して、コネクタが新しいテーブルを作成して構成する方法をより細かく制御できます。

allBQFieldsNullable
clusteringPartitionFieldNames
convertDoubleSpecialValues
partitionExpirationMs
sanitizeFieldNames
sanitizeTopics
timestampPartitionFieldName

これらのプロパティの詳細については、BigQuery Sink コネクタの構成をご覧ください。

Kafka メタデータ

kafkaDataFieldName フィールドと kafkaKeyFieldName フィールドをそれぞれ構成することで、メタデータ情報やキー情報などの Kafka の追加データを BigQuery テーブルにマッピングできます。メタデータ情報の例としては、Kafka トピック、パーティション、オフセット、挿入時間などがあります。

次のステップ

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

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