BigQuery サブスクリプションの作成

このドキュメントでは、BigQuery サブスクリプションの作成方法について説明します。 BigQuery サブスクリプションを作成するには、 Google Cloud コンソール、Google Cloud CLI、クライアント ライブラリ、 または Pub/Sub API を使用できます。

始める前に

このドキュメントを読む前に、次の内容をよく理解しておいてください。

BigQuery サブスクリプションを作成する前に、Pub/Sub と BigQuery に精通しているだけでなく、次の前提条件を満たしていることを確認してください。

  • BigQuery テーブルが存在している。あるいは、このドキュメントの後半のセクションで説明するように、BigQuery サブスクリプションの作成時に作成することもできます。

  • Pub/Sub トピックのスキーマと BigQuery テーブルの間の互換性。互換性のない BigQuery テーブルを追加すると、互換性に関連するエラー メッセージが出力されます。詳細については、スキーマの互換性をご覧ください。

必要なロールと権限

BigQuery サブスクリプションの作成に必要な権限を取得するには、管理者にプロジェクトに対するPub/Sub 編集者 （roles/pubsub.editor）IAM ロールを付与するよう依頼してください。ロールの付与については、プロジェクト、フォルダ、組織に対するアクセス権の管理をご覧ください。

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

必要な権限

BigQuery サブスクリプションを作成するには、次の権限が必要です。

  • pubsub.subscriptions.create プロジェクトに対する
  • pubsub.topics.attachSubscription トピックに対する

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

プロジェクト間のサブスクリプション

別のプロジェクトのトピックに対して 1 つのプロジェクトでサブスクリプションを作成する場合は、サブスクリプションを作成するプロジェクトに対する pubsub.subscriptions.create 権限と、トピックに対する pubsub.topics.attachSubscription 権限が必要です。

サービス アカウントに IAM のロールを付与する

Pub/Sub は、Identity and Access Management（IAM）サービス アカウントを使用して リソースにアクセスします Google Cloud 。デフォルトでは、 Pub/Sub サービス エージェント （service-PROJECT_NUMBER@gcp-sa-pubsub.iam.gserviceaccount.com）が使用されます。

Pub/Sub が BigQuery テーブルに書き込めるようにするには、サービス アカウントに BigQuery データ編集者 （roles/bigquery.dataEditor）ロールが必要です。サービス アカウントには、次のようにプロジェクトまたはテーブルの権限を付与できます。

プロジェクト

  1. コンソールで、[IAM] ページに移動します。 Google Cloud

    [IAM] に移動

  2. [Google 提供のロール付与を含む] を選択します。

  3. [Cloud Pub/Sub] サービス アカウントの行を見つけて、 [プリンシパルの編集] をクリックします。

  4. [別のロールを追加] をクリックし、[BigQuery データ編集者] ロールを選択します。

詳細については、 コンソールを使用して IAM ロールを付与するをご覧ください。

テーブル

  1. コンソールで、[BigQuery Studio] に移動します。 Google Cloud

    BigQuery Studio に移動

  2. [エクスプローラ] ペインの [名前とラベルでフィルタ] 検索ボックスに、 テーブルの名前を入力して [Enter] キーを押します。

  3. 検索結果で、権限を付与するテーブルの名前をクリックします。

  4. [**詳細**] タブで、 [**共有**] > [**権限の管理**] をクリックします。

  5. [Add principal] をクリックし、次の 形式でサービス アカウント識別子を入力します。

    service-PROJECT_NUMBER@gcp-sa-pubsub.iam.gserviceaccount.com。

  6. [ロールを割り当てる] リストで、[BigQuery データ編集者] を選択します。

  7. [保存] をクリックします。リソースのロールがプリンシパルに付与されます。

カスタム サービス アカウントを使用する

Cloud Pub/Sub サービス アカウントに BigQuery データ編集者 ロールを付与すると、プロジェクトでサブスクリプションを作成する権限を持つユーザーは誰でも BigQuery テーブルに書き込むことができます。より詳細な権限を付与する場合は、代わりに ユーザー管理のサービス アカウントを構成します。

ユーザー管理のサービス アカウントを構成して BigQuery に書き込むには、次の権限が必要です。

  • ユーザー管理のサービス アカウントに BigQuery データ編集者 ロールが必要です。

  • Cloud Pub/Sub サービス アカウントに、ユーザー管理のサービス アカウントに対する iam.serviceAccounts.getAccessToken 権限が必要です。

  • サブスクリプションを作成するユーザーに、ユーザー管理のサービス アカウントに対する iam.serviceAccounts.actAs 権限が必要です。

サブスクリプションを作成するときに、ユーザー管理のサービス アカウントを サブスクリプション サービス アカウントとして指定します。

BigQuery サブスクリプション プロパティ

BigQuery サブスクリプションは、一般的なサブスクリプション プロパティをすべてサポートしています。以降のセクションでは、BigQuery サブスクリプションに固有のプロパティについて説明します。

トピック スキーマを使用する

このオプションにより、Pub/Sub は、サブスクリプションが接続されている Pub/Sub トピックのスキーマを使用できます。さらに、Pub/Sub はメッセージ内のフィールドを BigQuery テーブル内の対応する列に書き込みます。

このオプションを使用する場合は、以下の追加要件を確認してください。

  • トピック スキーマ内のフィールドと BigQuery スキーマ内のフィールドは、同じ名前でなければならず、その型には相互に互換性が必要です。

  • トピック スキーマのオプション フィールドは BigQuery スキーマでも省略可能な必要があります。

  • トピック スキーマの必須フィールドは、BigQuery スキーマでは必要ありません。

  • BigQuery フィールドがトピック スキーマに存在しない場合、これらの BigQuery フィールドは NULLABLE モードにする必要があります。

  • トピック スキーマに、BigQuery スキーマに存在しない追加のフィールドがあり、このようなフィールドを削除できる場合は、[不明な項目を削除する] オプションを選択します。

  • サブスクリプション プロパティとして [トピック スキーマを使用する] または [テーブル スキーマを使用する] のいずれか 1 つのみを選択できます。

[トピック スキーマを使用する] または [テーブル スキーマを使用する] オプションを選択しない場合は、BigQuery テーブルに、BYTES、STRING、または JSON 型の data という列があることを確認してください。Pub/Sub は、この BigQuery 列にメッセージを書き込みます。

Pub/Sub トピック スキーマまたは BigQuery テーブル スキーマの変更は、BigQuery テーブルに書き込まれたメッセージにすぐに反映されない場合があります。たとえば、[不明なフィールドを削除する] オプションが有効で、フィールドが Pub/Sub スキーマには存在するが BigQuery スキーマにはない場合、BigQuery テーブルに書き込まれるメッセージには、BigQuery スキーマに追加した後も、フィールドが含まれていない場合があります。最終的に、スキーマが同期され、後続のメッセージにこのフィールドが含まれます。

BigQuery サブスクリプションで [テーブル スキーマを使用する] オプションを使用すると、BigQuery 変更データ キャプチャ（CDC）も利用できます。CDC は、既存の行を処理して変更を適用し、BigQuery テーブルを更新します。

この機能の詳細については、変更データ キャプチャを使用してテーブル更新をストリーミングするをご覧ください。

BigQuery サブスクリプションでこの機能を使用する方法については、BigQuery の変更データ キャプチャをご覧ください。

テーブル スキーマを使用する

このオプションにより、Pub/Sub は BigQuery テーブルのスキーマを使用して、JSON メッセージのフィールドを対応する列に書き込むことができます。このオプションを使用する場合は、以下の追加要件を確認してください。

  • BigQuery テーブルの各列の名前には、英字（a-z、A-Z）、数字（0-9）、アンダースコア（_）のみを使用できます。

  • 公開されるメッセージは JSON 形式である必要があります。

    BigQuery テーブルの列のデータ型が JSON の場合、Pub/Sub メッセージの対応するフィールドは、エスケープされた文字列で有効な JSON である必要があります。たとえば、myData という名前の列の場合、 メッセージ フィールドは "myData": "{\"key\":\"value\"}" にする必要があります。 有効な JSON が含まれていないメッセージは BigQuery によって拒否されます。

  • 次の JSON 変換がサポートされています。

    JSON 型 BigQuery のデータ型
    string NUMERIC、BIGNUMERIC、DATE、TIME、DATETIME、または TIMESTAMP
    number NUMERIC、BIGNUMERIC、DATE、TIME、DATETIME、または TIMESTAMP
    • number から DATE、DATETIME、TIME、または TIMESTAMP へのコンバージョンを使用する場合、数値は サポートされている表現に準拠する必要があります。
    • number から NUMERIC または BIGNUMERIC へのコンバージョンを使用する場合、値の適合率と範囲は、浮動小数点演算の IEEE 754 標準で許容されるものに制限されます。高い適合率やより広い範囲の値が必要な場合は、代わりに string から NUMERIC または BIGNUMERIC へのコンバージョンを使用します。
    • string から NUMERIC または BIGNUMERIC へのコンバージョンを使用する場合、Pub/Sub は文字列が人間が判読できる数値（"123.124" など）であると想定します。文字列を人間が判読できる数値として処理できない場合、Pub/Sub は文字列を BigDecimalByteStringEncoder でエンコードされたバイトとして扱います。

  • サブスクリプションのトピックにスキーマが関連付けられている場合は、メッセージ エンコード プロパティを JSON に設定する必要があります。

  • メッセージに存在しない BigQuery フィールドがある場合、これらの BigQuery フィールドは NULLABLE モードにする必要があります。

  • BigQuery スキーマに存在しない追加のフィールドがメッセージにあり、このようなフィールドを削除できる場合は、[不明なフィールドを削除する] オプションを選択します。

  • サブスクリプション プロパティとして [トピック スキーマを使用する] または [テーブル スキーマを使用する] のいずれか 1 つのみを選択できます。

[トピック スキーマを使用する] または [テーブル スキーマを使用する] オプションを選択しない場合は、BigQuery テーブルに、BYTES、STRING、または JSON 型の data という列があることを確認してください。Pub/Sub は、この BigQuery 列にメッセージを書き込みます。

BigQuery テーブル スキーマの変更は、BigQuery テーブルに書き込まれたメッセージにすぐに反映されない場合があります。 たとえば、[不明なフィールドを削除する] オプションが有効で、フィールドがメッセージには存在するが BigQuery スキーマにはない場合、BigQuery テーブルに書き込まれるメッセージには、BigQuery スキーマに追加した後も、フィールドが含まれていない場合があります。最終的に、スキーマが同期され、後続のメッセージにこのフィールドが含まれます。

BigQuery サブスクリプションで [テーブル スキーマを使用する] オプションを使用すると、BigQuery 変更データ キャプチャ（CDC）も利用できます。 CDC は、既存の行を処理して変更を適用し、BigQuery テーブルを更新します。

この機能の詳細については、変更データ キャプチャを使用してテーブル更新をストリーミングするをご覧ください。

BigQuery サブスクリプションでこの機能を使用する方法については、BigQuery 変更データ キャプチャをご覧ください。

不明な項目を削除する

このオプションは、[トピック スキーマを使用する] または [テーブル スキーマを使用する] オプションで使用します。有効にすると、このオプションを使用すると、Pub/Sub は、トピック スキーマやメッセージには存在するが BigQuery スキーマには存在しないフィールドをドロップできます。BigQuery スキーマの一部ではないフィールドは、BigQuery テーブルにメッセージを書き込むときに削除されます。

[不明な項目を削除する] が設定されていない場合、余分な項目を含むメッセージは BigQuery に書き込まれず、サブスクリプション バックログに残ります。 デッドレター トピックを構成しない限り、デッドレター トピック を構成しない限り、サブスクリプション バックログに残ります。

[不明な項目を削除する] 設定は、Pub/Sub トピック スキーマまたは BigQuery テーブル スキーマのいずれにも定義されていないフィールドには影響しません。この場合、有効な Pub/Sub メッセージがサブスクリプションに配信されます。ただし、BigQuery にはこれらの追加フィールド用に定義された列がないため、これらのフィールドは BigQuery の書き込みプロセス中に削除されます。この動作を防ぐには、Pub/Sub メッセージに含まれるフィールドが BigQuery テーブル スキーマにも含まれていることを確認します。

追加フィールドに関する動作は、使用する特定のスキーマタイプ（Avro、Protocol Buffer）とエンコード（JSON、バイナリ）によっても異なります。これらの要素が追加フィールドの処理に与える影響については、特定のスキーマタイプとエンコードのドキュメントをご覧ください。

メタデータを書き込む

このオプションを使用すると、Pub/Sub が各メッセージのメタデータを BigQuery テーブルの追加の列に書き込むようにする場合は、このオプションを選択できます。それ以外の場合、メタデータは BigQuery テーブルに書き込まれません。

[メタデータを書き込む] オプションを選択する場合は、BigQuery テーブルに次の表に示すフィールドがあることを確認してください。

メタデータを書き込むオプションを選択しない場合、use_topic_schema が true でない限り、宛先 BigQuery テーブルでは data フィールドのみが必要になります。[メタデータを書き込む] と [トピック スキーマを使用する] の両方のオプションを選択した場合、トピックのスキーマには、次の名前にメタデータ パラメータと一致する名前のフィールドを含めないでください。この制限には、スネークケース パラメータのキャメルケース バージョンが含まれます。

パラメータ
subscription_name

STRING

サブスクリプションの名前。
message_id

STRING

メッセージの ID
publish_time

TIMESTAMP

メッセージのパブリッシュ時刻。
data

BYTES、STRING、または JSON

メッセージの本文。

data フィールドは、 [**トピック スキーマを使用する**] または [**テーブル スキーマを使用する**] を選択しないすべての宛先 BigQuery テーブルに必要です。フィールドの型が JSON の場合、メッセージ本文を有効な JSON にする必要があります。
attributes

STRING または JSON

すべてのメッセージ属性を含む JSON オブジェクト。また、存在する場合、順序指定キーなど、 Pub/Sub メッセージの一部である 追加のフィールドも含まれています。

サービス アカウント

BigQuery テーブルにメッセージを書き込むには、次のオプションがあります。

  • カスタム サービス アカウントを構成して、サービス アカウントに対する iam.serviceAccounts.actAs 権限を持つユーザーのみが、テーブルに書き込むサブスクリプションを作成できるようにします。iam.serviceAccounts.actAs 権限を含むロールの例として、サービス アカウント ユーザー （roles/iam.serviceAccountUser）ロールがあります。

  • プロジェクトでサブスクリプションを作成できるユーザーがテーブルに書き込むサブスクリプションを作成できるようにする、デフォルトの Pub/Sub サービス エージェントを使用します。カスタム サービス アカウントを指定しない場合、Pub/Sub サービス エージェントがデフォルト設定になります。

BigQuery サブスクリプションの作成

BigQuery 配信でサブスクリプションを作成する手順は次のとおりです。

コンソール

  1. コンソールで、[サブスクリプションの作成] ページに移動します。 Google Cloud

    [サブスクリプション] に移動

  2. [サブスクリプション ID] フィールドに名前を入力します。サブスクリプションの指定方法については、 トピックまたはサブスクリプションの指定方法のガイドラインをご覧ください。

  3. [Cloud Pub/Sub トピックを選択してください] ボックスに、メッセージを受信するトピックを入力または選択します。

  4. [**配信タイプ**] で [**BigQuery への書き込み**] を選択します。

  5. BigQuery テーブルを選択します。

    1. [**プロジェクト**] で、BigQuery テーブルを含む Google Cloud プロジェクトを選択します。

    2. [データセット] で、既存のデータセットを選択するか、 [新しいデータセットの作成] をクリックして新しいデータセットを作成します。データセットの作成については、データセットの作成をご覧ください。

    3. [テーブル] フィールドにテーブルの名前を入力します。新しいテーブルを作成するには、BigQuery の [新しいテーブルを作成] ページに移動するリンクをクリックします。このページは別のタブで開きます。テーブルの作成については、 テーブルの作成と使用をご覧ください。

  6. [スキーマ構成] で、次のいずれかのオプションを選択します。

    • スキーマを使用しない。Pub/Sub は、メッセージのバイトを data という名前の列に書き込みます。

    • トピック スキーマを使用する 。Pub/Sub は、トピックに関連付けられているスキーマを使用します。詳細については、 トピック スキーマを使用するをご覧ください。

    • テーブル スキーマを使用する 。Pub/Sub は BigQuery テーブルのスキーマを使用します。詳細については、 テーブル スキーマを使用するをご覧ください。

  7. 省略可。メッセージ メタデータを BigQuery テーブルに書き込むには、[メタデータを書き込む] を選択します。詳細については、 メタデータを書き込むをご覧ください。

  8. 省略可。BigQuery テーブル スキーマに存在しないフィールドを削除するには、[不明な項目を削除する] を選択します。詳細については、不明な項目を削除するをご覧ください。

  9. 必要に応じて、 一般的なサブスクリプション プロパティ を構成します。メッセージ エラーを処理するには、デッドレター を有効にすることを強くおすすめします。詳細については、 デッドレター トピックをご覧ください。

  10. [作成] をクリックします。

gcloud

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

    Cloud Shell をアクティブにする

    コンソールの下部にある Google Cloud Cloud Shell セッションが開始し、コマンドライン プロンプトが表示されます。Cloud Shell はシェル環境です 。Google Cloud CLI がすでにインストールされており、現在のプロジェクトの値もすでに設定されています 。セッションが初期化されるまで数秒かかることがあります。

  2. Pub/Sub サブスクリプションを作成するには、gcloud pubsub subscriptions create コマンドを使用します。

    gcloud pubsub subscriptions create SUBSCRIPTION_ID \
    --topic=TOPIC_ID \
    --bigquery-table=PROJECT_ID.DATASET_ID.TABLE_ID

    カスタム サービス アカウントを使用する場合は、追加の引数として指定します。

    gcloud pubsub subscriptions create SUBSCRIPTION_ID \
    --topic=TOPIC_ID \
    --bigquery-table=PROJECT_ID.DATASET_ID.TABLE_ID \
    --bigquery-service-account-email=SERVICE_ACCOUNT_NAME

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

    • SUBSCRIPTION_ID: サブスクリプションの ID を指定します。
    • TOPIC_ID: トピックの ID を指定します。このトピックでは、スキーマが必要です。
    • PROJECT_ID: プロジェクトの ID を指定します。
    • DATASET_ID: 既存のデータセットの ID を指定します。データセットを作成するには、 データセットの作成をご覧ください。
    • TABLE_ID: 既存のテーブルの ID を指定します。トピックにスキーマがない場合、このテーブルには data フィールドが必要です。テーブルを作成するには、スキーマ定義を含む空のテーブルを作成するをご覧ください。
    • SERVICE_ACCOUNT_NAME: BigQuery への書き込みに使用するサービス アカウントの名前を指定します。

C++

このサンプルを試す前に、 クイックスタート: クライアント ライブラリの使用の C++ の設定手順を実施してください。 詳細については、Pub/Sub C++ API リファレンス ドキュメントをご覧ください。 

namespace pubsub = ::google::cloud::pubsub;
namespace pubsub_admin = ::google::cloud::pubsub_admin;
[](pubsub_admin::SubscriptionAdminClient client,
   std::string const& project_id, std::string const& topic_id,
   std::string const& subscription_id, std::string const& table_id) {
  google::pubsub::v1::Subscription request;
  request.set_name(
      pubsub::Subscription(project_id, subscription_id).FullName());
  request.set_topic(pubsub::Topic(project_id, topic_id).FullName());
  request.mutable_bigquery_config()->set_table(table_id);
  auto sub = client.CreateSubscription(request);
  if (!sub) {
    if (sub.status().code() == google::cloud::StatusCode::kAlreadyExists) {
      std::cout << "The subscription already exists\n";
      return;
    }
    throw std::move(sub).status();
  }

  std::cout << "The subscription was successfully created: "
            << sub->DebugString() << "\n";
}

C#

このサンプルを試す前に、 クイックスタート: クライアント ライブラリの使用の C# の設定手順を実施してください。 詳細については、Pub/Sub C# API リファレンス ドキュメントをご覧ください。 


using Google.Cloud.PubSub.V1;

public class CreateBigQuerySubscriptionSample
{
    public Subscription CreateBigQuerySubscription(string projectId, string topicId, string subscriptionId, string bigqueryTableId)
    {
        SubscriberServiceApiClient subscriber = SubscriberServiceApiClient.Create();
        TopicName topicName = TopicName.FromProjectTopic(projectId, topicId);
        SubscriptionName subscriptionName = SubscriptionName.FromProjectSubscription(projectId, subscriptionId);

        var subscriptionRequest = new Subscription
        {
            SubscriptionName = subscriptionName,
            TopicAsTopicName = topicName,
            BigqueryConfig = new BigQueryConfig
            {
                Table = bigqueryTableId
            }
        };
        var subscription = subscriber.CreateSubscription(subscriptionRequest);
        return subscription;
    }
}

Go

次のサンプルでは、Go Pub/Sub クライアント ライブラリのメジャー バージョン（v2）を使用しています。v1 ライブラリをまだ使用している場合は、 v2 への移行ガイドをご覧ください。 v1 のコードサンプルのリストについては、 非推奨のコードサンプルをご覧ください。

このサンプルを試す前に、 クイックスタート: クライアント ライブラリの使用の Go の設定手順を実施してください。 詳細については、Pub/Sub Go API のリファレンス ドキュメントをご覧ください。

import (
	"context"
	"fmt"
	"io"

	"cloud.google.com/go/pubsub/v2"
	"cloud.google.com/go/pubsub/v2/apiv1/pubsubpb"
)

// createBigQuerySubscription creates a Pub/Sub subscription that exports messages to BigQuery.
func createBigQuerySubscription(w io.Writer, projectID, topic, subscription, table string) error {
	// projectID := "my-project"
	// topic := "projects/my-project-id/topics/my-topic"
	// subscription := "projects/my-project/subscriptions/my-sub"
	// table := "my-project-id.dataset_id.table_id"
	ctx := context.Background()
	client, err := pubsub.NewClient(ctx, projectID)
	if err != nil {
		return fmt.Errorf("pubsub.NewClient: %w", err)
	}
	defer client.Close()

	sub, err := client.SubscriptionAdminClient.CreateSubscription(ctx, &pubsubpb.Subscription{
		Name:  subscription,
		Topic: topic,
		BigqueryConfig: &pubsubpb.BigQueryConfig{
			Table:         table,
			WriteMetadata: true,
		},
	})
	if err != nil {
		return fmt.Errorf("failed to create subscription: %w", err)
	}
	fmt.Fprintf(w, "Created BigQuery subscription: %v\n", sub)

	return nil
}

Java

このサンプルを試す前に、 クイックスタート: クライアント ライブラリの使用の Java の設定手順を実施してください。 詳細については、Pub/Sub Java API リファレンス ドキュメントをご覧ください。 

import com.google.cloud.pubsub.v1.SubscriptionAdminClient;
import com.google.pubsub.v1.BigQueryConfig;
import com.google.pubsub.v1.ProjectSubscriptionName;
import com.google.pubsub.v1.ProjectTopicName;
import com.google.pubsub.v1.Subscription;
import java.io.IOException;

public class CreateBigQuerySubscriptionExample {
  public static void main(String... args) throws Exception {
    // TODO(developer): Replace these variables before running the sample.
    String projectId = "your-project-id";
    String topicId = "your-topic-id";
    String subscriptionId = "your-subscription-id";
    String bigqueryTableId = "your-project.your-dataset.your-table";

    createBigQuerySubscription(projectId, topicId, subscriptionId,