変更ストリームを作成して管理する

このページでは、GoogleSQL 言語データベースと PostgreSQL 言語データベースの Spanner 変更ストリームを作成、変更、表示する方法について説明します。変更ストリームの詳細については、変更ストリームについてをご覧ください。

変更ストリームはスキーマ オブジェクトであるため、他の種類のデータベース定義作業（データセットの作成やインデックスの追加など）に使用されるものと同じ DDL によるスキーマ更新を使用して作成、管理します。

Spanner は、スキーマを変更する DDL ステートメント（変更ストリームの作成、変更、削除に使用されたステートメントを含む）を送信すると、長時間実行オペレーションを開始します。新規または変更された変更ストリームは、この長時間実行オペレーションの完了時に、新しい構成で指定された列やテーブルのモニタリングを開始します。

変更ストリームを作成

変更ストリームを作成するには、その名前と監視するスキーマ オブジェクト（データベース全体か、特定のテーブルと列のリスト）を指定する必要があります。必要に応じて、次のいずれかを使用して変更ストリームを構成できます。

• データの保持期間を指定して、デフォルトの 1 日の保持期間をオーバーライドします。
• デフォルト値のキャプチャ タイプ OLD_AND_NEW_VALUES をオーバーライドする値キャプチャ タイプを指定します。
• TTL ベースの削除フィルタを適用して、変更ストリームから TTL ベースの削除を除外します。
• テーブル変更フィルタを適用して、INSERT、UPDATE、DELETE のすべてのテーブルの変更を除外します。

• トランザクション レベルのレコード除外を有効にすると、特定のトランザクションを変更ストリームから除外できます。

CREATE CHANGE STREAM CHANGE_STREAM_NAME
  [FOR column_or_table_watching_definition[, ... ] ]
  [
    OPTIONS (
      retention_period = timespan,
      value_capture_type = type,
      exclude_ttl_deletes = boolean,
      exclude_insert = boolean,
      exclude_update = boolean,
      exclude_delete = boolean,
      allow_txn_exclusion = boolean
    )
  ]

CREATE CHANGE STREAM CHANGE_STREAM_NAME
  [FOR column_or_table_watching_definition[, ... ] ]
  [
    WITH (
      retention_period = timespan,
      value_capture_type = type,
      exclude_ttl_deletes = boolean,
      exclude_insert = boolean,
      exclude_update = boolean,
      exclude_delete = boolean,
      allow_txn_exclusion = boolean
    )
  ]

新しい変更ストリームを作成した長時間実行オペレーションが完了すると直ちに、この変更ストリームは割り当てられたスキーマ オブジェクトのモニタリングを開始します。

次の例では、さまざまな構成で変更ストリームを作成する方法を示します。

データベース全体を監視する

データベースのテーブル全体で行われるすべてのデータ変更を監視する変更ストリームを作成するには、ALL キーワードを使用します。

CREATE CHANGE STREAM EverythingStream
FOR ALL;

ALL 構成では、今後作成されるデータべースのデータテーブルと列がすべて、すぐに暗黙的に含まれます。ビュー、情報スキーマ テーブル、通常のデータテーブル以外の他のオブジェクトは含まれません。

特定のテーブルをモニタリングする

変更ストリームのスコープをデータベース全体ではなく特定のテーブルに制限するには、1 つ以上のテーブルのリストを指定します。

CREATE CHANGE STREAM SingerAlbumStream
FOR Singers, Albums;

Spanner は、テーブル全体をモニタリングする変更ストリームを自動的に更新し、列の追加や削除など、テーブルに影響するスキーマ変更を反映します。

特定の列を監視する

table(column_1[, column_2, ...]) 構文を使用して、指定したテーブル内にある 1 つ以上の特定の非キー列に対する変更を監視します。

CREATE CHANGE STREAM NamesAndTitles
FOR Singers(FirstName, LastName), Albums(Title);

すべての変更ストリームは、監視対象のすべてのテーブルの主キーを常時追跡するため、主キー列を指定することはできません。これにより、すべてのデータ変更レコードで、変更された行を主キーで識別できます。

テーブルと列を 1 つのストリームで監視する

前の 2 つの例のテーブル監視構文と列監視構文は、1 つの変更ストリームで組み合わせることが可能です。

CREATE CHANGE STREAM NamesAndAlbums
FOR Singers(FirstName, LastName), Albums;

長い保持期間を指定する

変更ストリーム データの保持期間をデフォルトの 1 日よりも長く指定するには、retention_period を最大 30 日の期間（時間（h）または日（d））に設定します。

2 つの例を以下に示します。

CREATE CHANGE STREAM LongerDataRetention
FOR ALL
OPTIONS ( retention_period = '36h' );

CREATE CHANGE STREAM MaximumDataRetention
FOR ALL
OPTIONS ( retention_period = '7d' );

CREATE CHANGE STREAM LongerDataRetention
FOR ALL
WITH ( retention_period = '36h' );

CREATE CHANGE STREAM MaximumDataRetention
FOR ALL
WITH ( retention_period = '7d' );

別の値キャプチャ タイプを指定する

OLD_AND_NEW_VALUES 以外の変更ストリーム値キャプチャ タイプを指定するには、value_capture_type を NEW_VALUES または NEW_ROW に設定します。次に例を示します。

CREATE CHANGE STREAM NewRowChangeStream
FOR ALL
OPTIONS ( value_capture_type = 'NEW_ROW' );

CREATE CHANGE STREAM NewValuesChangeStream
FOR ALL
OPTIONS ( value_capture_type = 'NEW_VALUES' );

CREATE CHANGE STREAM NewRowChangeStream
FOR ALL
WITH ( value_capture_type = 'NEW_ROW' );

CREATE CHANGE STREAM NewValuesChangeStream
FOR ALL
WITH ( value_capture_type = 'NEW_VALUES' );

TTL ベースの削除をフィルタ

exclude_ttl_deletes フィルタを使用して、変更ストリームのスコープで TTL ベースの削除をフィルタできます。

このフィルタの仕組みの詳細については、有効期間ベースの削除フィルタをご覧ください。

CREATE CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
OPTIONS (exclude_ttl_deletes = true)

CREATE CHANGE STREAM NewFilterChangeStream FOR ALL
OPTIONS (exclude_ttl_deletes = true)

CREATE CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
WITH (exclude_ttl_deletes = true)

CREATE CHANGE STREAM NewFilterChangeStream FOR ALL
WITH (exclude_ttl_deletes = true)

既存の変更ストリームに TTL ベースの削除フィルタを追加または削除するには、TTL ベースの削除フィルタを変更するをご覧ください。変更ストリームのフィルタは、変更ストリームの定義を DDL として表示することで確認できます。

テーブル変更タイプでのフィルタ

使用可能な次のフィルタ オプションを使用して、変更ストリームのスコープでこれらのテーブル変更の 1 つ以上をフィルタします。

• exclude_insert: INSERT テーブルのすべての変更を除外します。
• exclude_update: UPDATE テーブルのすべての変更を除外します。
• exclude_delete: DELETE テーブルのすべての変更を除外します。

これらのフィルタの仕組みの詳細については、テーブル変更タイプのフィルタをご覧ください。

CREATE CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
OPTIONS (MOD_TYPE_FILTER_NAME = true)

CREATE CHANGE STREAM NewFilterChangeStream FOR ALL
OPTIONS (exclude_insert = true, exclude_update = true)

CREATE CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
WITH (MOD_TYPE_FILTER_NAME = true)

CREATE CHANGE STREAM NewFilterChangeStream FOR ALL
WITH (exclude_insert = true, exclude_update = true)

既存の変更ストリームにテーブル変更タイプのフィルタを追加または削除するには、テーブル変更タイプでのフィルタを変更するをご覧ください。変更ストリームに存在するテーブル変更タイプのフィルタは、変更ストリームの定義を DDL として表示することで確認できます。

トランザクション レベルのレコードの除外を有効にする

変更ストリームを有効にして、指定した書き込みトランザクションからレコードを除外するには、変更ストリームを作成するときに allow_txn_exclusion オプションを設定するか、既存の変更ストリームを変更します。

このオプションの仕組みの詳細については、トランザクション レベルのレコード除外をご覧ください。

CREATE CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
OPTIONS (allow_txn_exclusion = true)

CREATE CHANGE STREAM NewChangeStream FOR ALL
OPTIONS (allow_txn_exclusion = true)

CREATE CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
WITH (allow_txn_exclusion = true)

CREATE CHANGE STREAM NewChangeStream FOR ALL
WITH (allow_txn_exclusion = true)

既存の変更ストリームからのトランザクション レベルのレコードの除外を有効または無効にするには、トランザクション レベルのレコードの除外を変更するをご覧ください。このオプションの設定を確認するには、変更ストリームの定義を DDL として表示するをご覧ください。

変更ストリームから除外する書き込みトランザクションを指定する

変更ストリームから除外する書き込みトランザクションを指定するには、exclude_txn_from_change_streams パラメータを true に設定する必要があります。次のコードサンプルは、クライアント ライブラリを使用して変更ストリームから除外する書き込みトランザクションを指定する方法を示しています。

import (
	"context"
	"fmt"
	"io"

	"cloud.google.com/go/spanner"
)

// readWriteTxnExcludedFromChangeStreams executes the insert and update DMLs on
// Singers table excluded from tracking change streams with ddl option
// allow_txn_exclusion = true.
func readWriteTxnExcludedFromChangeStreams(w io.Writer, db string) error {
	// db = `projects/<project>/instances/<instance-id>/database/<database-id>`
	ctx := context.Background()
	client, err := spanner.NewClient(ctx, db)
	if err != nil {
		return fmt.Errorf("readWriteTxnExcludedFromChangeStreams.NewClient: %w", err)
	}
	defer client.Close()

	_, err = client.ReadWriteTransactionWithOptions(ctx, func(ctx context.Context, txn *spanner.ReadWriteTransaction) error {
		stmt := spanner.Statement{
			SQL: `INSERT Singers (SingerId, FirstName, LastName)
					VALUES (111, 'Virginia', 'Watson')`,
		}
		_, err := txn.Update(ctx, stmt)
		if err != nil {
			return fmt.Errorf("readWriteTxnExcludedFromChangeStreams.Update: %w", err)
		}
		fmt.Fprintln(w, "New singer inserted.")
		stmt = spanner.Statement{
			SQL: `UPDATE Singers SET FirstName = 'Hi' WHERE SingerId = 111`,
		}
		_, err = txn.Update(ctx, stmt)
		if err != nil {
			return fmt.Errorf("readWriteTxnExcludedFromChangeStreams.Update: %w", err)
		}
		fmt.Fprint(w, "Singer first name updated.")
		return nil
	}, spanner.TransactionOptions{ExcludeTxnFromChangeStreams: true})
	if err != nil {
		return err
	}
	return nil
}

static void readWriteTxnExcludedFromChangeStreams(DatabaseClient client) {
  // Exclude the transaction from allowed tracking change streams with alloww_txn_exclusion=true.
  // This exclusion will be applied to all the individual operations inside this transaction.
  client
      .readWriteTransaction(Options.excludeTxnFromChangeStreams())
      .run(
          transaction -> {
            transaction.executeUpdate(
                Statement.of(
                    "INSERT Singers (SingerId, FirstName, LastName)\n"
                        + "VALUES (1341, 'Virginia', 'Watson')"));
            System.out.println("New singer inserted.");

            transaction.executeUpdate(
                Statement.of("UPDATE Singers SET FirstName = 'Hi' WHERE SingerId = 111"));
            System.out.println("Singer first name updated.");

            return null;
          });
}

変更ストリームを変更する

変更ストリームの構成を変更するには、ALTER CHANGE STREAM DDL ステートメントを使用します。CREATE CHANGE STREAM と同様の構文を使用します。ストリームの監視対象の列やデータ保持期間の長さは変更できます。データ変更レコードを保持しながら、監視を全体で一時停止することもできます。

変更ストリームの監視対象を変更する

この例では、Songs テーブル全体を以前に構成した NamesAndAlbums 変更ストリームに追加します。

ALTER CHANGE STREAM NamesAndAlbums
SET FOR Singers(FirstName, LastName), Albums, Songs;

Spanner は、データベースのスキーマ内の変更ストリームの定義を更新する長時間実行オペレーションの完了時に、名前付き変更ストリームの動作を新しい構成に置き換えます。

変更ストリームのデータ保持期間を変更する

変更ストリームが内部レコードを保持する時間を変更するには、ALTER CHANGE STREAM DDL ステートメントで retention_period を設定します。

この例では、データ保持期間を、前に作成した NamesAndAlbums 変更ストリームに合わせて調整します。

ALTER CHANGE STREAM NamesAndAlbums
SET OPTIONS ( retention_period = '36h' );

ALTER CHANGE STREAM NamesAndAlbums
SET ( retention_period = '36h' );

変更ストリームの値キャプチャ タイプを変更する

変更ストリームの値キャプチャ タイプを変更するには、ALTER CHANGE STREAM DDL ステートメントで value_capture_type 句を設定します。

この例では、値キャプチャ タイプを NEW_VALUES に調整しています。

ALTER CHANGE STREAM NamesAndAlbums
SET OPTIONS ( value_capture_type = 'NEW_VALUES' );

ALTER CHANGE STREAM NamesAndAlbums
SET ( value_capture_type = 'NEW_VALUES' );

TTL ベースの削除フィルタを変更する

変更ストリームの TTL ベースの削除フィルタを変更するには、ALTER CHANGE STREAM DDL ステートメントで exclude_ttl_deletes フィルタを設定します。これを使用して、既存の変更ストリームにフィルタを追加することや、既存の変更ストリームからフィルタを削除することが可能です。

これらのフィルタの仕組みの詳細については、有効期間ベースの削除フィルタをご覧ください。

既存の変更ストリームに TTL ベースの削除フィルタを追加する

ALTER CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
SET OPTIONS (exclude_ttl_deletes = true)

ALTER CHANGE STREAM NewFilterChangeStream FOR ALL
SET OPTIONS (exclude_ttl_deletes = true)

ALTER CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
SET (exclude_ttl_deletes = true)

ALTER CHANGE STREAM NewFilterChangeStream FOR ALL
SET (exclude_ttl_deletes = true)

既存の変更ストリームから TTL ベースの削除フィルタを削除する

ALTER CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
SET OPTIONS (exclude_ttl_deletes = false)

ALTER CHANGE STREAM NewFilterChangeStream FOR ALL
SET OPTIONS (exclude_ttl_deletes = false)

ALTER CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
SET (exclude_ttl_deletes = false)

ALTER CHANGE STREAM NewFilterChangeStream FOR ALL
SET (exclude_ttl_deletes = false)

テーブル変更タイプでのフィルタを変更する

変更ストリームのテーブル変更タイプのフィルタを変更するには、ALTER CHANGE STREAM DDL ステートメントでフィルタタイプを設定します。これを使用して、変更ストリームに新しいフィルタを追加することや、変更ストリームから既存のフィルタを削除することが可能です。

既存の変更ストリームにテーブル変更タイプのフィルタを追加する

ALTER CHANGE STREAM CHANGE_STREAM_NAME
SET OPTIONS (MOD_TYPE_FILTER_NAME = true)

ALTER CHANGE STREAM NewFilterChangeStream
SET OPTIONS (exclude_delete = true)

ALTER CHANGE STREAM CHANGE_STREAM_NAME
SET (MOD_TYPE_FILTER_NAME = true)

ALTER CHANGE STREAM NewFilterChangeStream
SET (exclude_delete = true)

既存の変更ストリームからテーブル変更タイプのフィルタを削除する

ALTER CHANGE STREAM CHANGE_STREAM_NAME
SET OPTIONS (MOD_TYPE_FILTER_NAME = false)

ALTER CHANGE STREAM NewFilterChangeStream
SET OPTIONS (exclude_delete = false)

ALTER CHANGE STREAM CHANGE_STREAM_NAME
SET (MOD_TYPE_FILTER_NAME = false)

ALTER CHANGE STREAM NewFilterChangeStream
SET (exclude_delete = false)

トランザクション レベルのレコードの除外を許可するように変更ストリームを変更する

指定した書き込みトランザクションからレコードを除外するように変更ストリームを変更できます。これを行うには、ALTER CHANGE STREAM DDL ステートメントで allow_txn_exclusion オプションを true に設定します。このオプションを設定しないか、false に設定した場合、変更ストリームはすべての書き込みトランザクションを監視します。

このオプションの仕組みの詳細については、トランザクション レベルのレコード除外をご覧ください。

既存の変更ストリームでトランザクション レベルのレコード除外を有効にする

ALTER CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
SET OPTIONS (allow_txn_exclusion = true)

ALTER CHANGE STREAM NewAllowedChangeStream FOR ALL
SET OPTIONS (allow_txn_exclusion = true)

ALTER CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
SET (allow_txn_exclusion = true)

ALTER CHANGE STREAM NewAllowedChangeStream FOR ALL
SET (allow_txn_exclusion = true)

既存の変更ストリームでトランザクション レベルのレコード除外を無効にする

ALTER CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
SET OPTIONS (allow_txn_exclusion = false)

ALTER CHANGE STREAM NewFilterChangeStream FOR ALL
SET OPTIONS (allow_txn_exclusion = false)

ALTER CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
SET (allow_txn_exclusion = false)

ALTER CHANGE STREAM NewAllowedChangeStream FOR ALL
SET (allow_txn_exclusion = false)

変更ストリームを一時停止する

変更ストリームがアクティビティを停止しても、少なくともデータ保持期間中は内部レコードを保持したい場合は、変更して何も監視しないように変更できます。

これを行うには、ALTER CHANGE STREAM DDL ステートメントを発行して変更ストリームの定義を特別なフレーズ DROP FOR ALL に置き換えます。例:

ALTER CHANGE STREAM MyStream DROP FOR ALL;

ストリームはデータベースに引き続き存在するものの、オブジェクトの監視は行わず、データ変更レコードも生成しません。既存の変更レコードは、ストリームのデータ保持ポリシーに沿ってそのまま残ります。

一時停止したストリームを再開するには、以前の構成で別の ALTER CHANGE STREAM ステートメントを発行します。

変更ストリームを削除する

変更ストリームを完全に削除するには、ストリームの名前を指定して DROP CHANGE STREAM ステートメントを発行します。

DROP CHANGE STREAM NamesAndAlbums;

Spanner はストリームを直ちに停止し、データベースのスキーマから削除してデータ変更レコードを削除します。

変更ストリームのリスト化と表示

Google Cloud コンソールには、データベースの変更ストリーム定義をリスト化し確認するためのウェブ インターフェースが用意されています。変更ストリームの構造は、同等の DDL ステートメントとして表示するか、データベースの情報スキーマをクエリして表示することもできます。

Google Cloud コンソールで変更ストリームを表示する

データベースの変更ストリームのリストを表示してその定義を確認するには:

1. Google Cloud コンソールの [Spanner インスタンス] ページに移動します。

   [インスタンス] ページを開く

2. 該当するインスタンスとデータベースに移動します。

3. ナビゲーション メニューで [変更ストリーム] をクリックします。

これにより、そのデータベースのすべての変更ストリームのリストと、それぞれの構成の要約が表示されます。ストリーム名をクリックすると、監視対象のテーブルと列の詳細が表示されます。

DDL として変更ストリームの定義を表示する

データベースのスキーマを DDL として表示すると、すべての変更ストリームの説明が含まれ、それらが CREATE CHANGE STREAM ステートメントとして表示されます。

• コンソールからこれを行うには、 Google Cloud コンソールのデータベース ページで [同等の DDL を表示] リンクをクリックします。

• コマンドラインからこれを行うには、Google Cloud CLI の ddl describe コマンドを使用します。

変更ストリームに関する情報スキーマをクエリ検索する

変更ストリームについては、データベースの情報スキーマを直接クエリ検索できます。次の表に、変更ストリームの名前、モニタリング対象のテーブルと列、保持期間を定義するメタデータを示します。

• information_schema.change_streams

• information_schema.change_stream_tables

• information_schema.change_stream_columns

• information_schema.change_stream_options

変更ストリームのベスト プラクティス

変更ストリームを構成して管理する際のベスト プラクティスを以下にいくつか紹介します。

別のメタデータ データベースについて検討する

変更ストリームでは、メタデータ データベースを使用して内部状態が維持されます。メタデータ データベースは、変更ストリームを含むデータベースと同じ、または異なる場合のいずれも問題ありません。メタデータ ストレージ用に個別のデータベースを作成することをおすすめします。

Spanner 変更ストリーム コネクタには、メタデータ データベースに対する読み取り / 書き込み権限が必要です。このデータベースをスキーマとともに準備する必要はありません。コネクタがそれに対処します。

別のメタデータ データベースを使用すると、コネクタがアプリケーション データベースに直接書き込むことを許可することによって発生する可能性のある複雑さを排除できます。

• メタデータ データベースを、変更ストリームを備えた本番環境のデータベースから分離すると、コネクタには、本番環境のデータベースに対する読み取り権限のみが必要になります。

• コネクタのトラフィックを別のメタデータ データベースに制限することで、コネクタ自体によって実行される書き込みは本番環境の変更ストリームに含まれません。これは、データベース全体をモニタリングする変更ストリームに特に関係します。

メタデータの保存に別のデータベースを使用していない場合は、インスタンス上の変更ストリーム コネクタの CPU の影響をモニタリングすることをおすすめします。

新しい変更ストリームをベンチマークし、必要に応じてサイズを変更する

本番環境インスタンスに新しい変更ストリームを追加する前に、変更ストリームを有効にしてステージング インスタンスで現実的なワークロードのベンチマークを実施することを検討してください。これにより、コンピューティング容量とストレージ容量を増やすためにインスタンスにノードを追加する必要があるかどうかを判断できます。

CPU とストレージの指標が安定するまで、これらのテストを実行します。最適な状態としては、インスタンスの CPU 使用率が推奨最大値を下回り、ストレージ使用量がインスタンスのストレージ上限を超えないようにする必要があります。

複数のリージョンを使用して負荷分散を行う

マルチリージョン インスタンス構成で変更ストリームを使用する場合は、デフォルトのリーダー リージョンとは異なるリージョンで処理パイプラインを実行することを検討してください。これにより、リーダー以外のレプリカ間でストリーミングの読み込みを分散処理できます。ただし、ロード バランシングよりもストリーミングの遅延を可能な限り低減することを優先する必要がある場合は、リーダー リージョンでストリーミングの読み込みを実行します。

次のステップ

• Dataflow を使用して変更ストリーム パイプラインを構築する。

• テンプレートを使用して、変更ストリームを BigQuery、Cloud Storage、Pub/Sub へ速やかに接続する。