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

イベントとストリーム

データストリームには、次のデータ階層があります。

ストリーム。データのソースと宛先で構成されます。
オブジェクト。特定のデータベースのテーブルなど、ストリームの一部です。
イベント。データベースの挿入など、特定のオブジェクトによって生成される 1 つの変更です。

ストリーム、オブジェクト、イベントには、データとメタデータが関連付けられています。このデータとメタデータは、さまざまな目的で使用できます。

イベントについて

各イベントは、次の 3 種類のデータで構成されています。

イベントデータ: ストリーム元から生じたオブジェクトからのデータ自体の変更を表します。各イベントには、変更された行全体が含まれます。
一般的なメタデータ: このメタデータは、Datastream によって生成されたすべてのイベントに表示されます。これは、宛先での重複データの削除などのアクションに使用されます。
ソース固有のメタデータ: このメタデータは、特定のストリームソースによって生成されたすべてのイベントに表示されます。このメタデータは、ソースによって異なります。

イベントデータ

イベントデータは、ストリーム元から生じた特定のオブジェクトからの変更すべてのペイロードです。

イベントは Avro 形式または JSON 形式です。

Avro 形式を使用する場合、イベントには、各列ごとに列インデックスと値が含まれます。列インデックスを使用して、Avro ヘッダーのスキーマから列名、統合タイプを取得できます。

JSON 形式を使用する場合、イベントには、各列ごとに列名と値が含まれます。

イベントメタデータは、イベントの発生元に関する情報の収集だけでなく、宛先での重複データの削除や、下流の利用者によるイベントの並べ替えにも使用できます。

以下の表では、汎用およびソース固有のイベントメタデータのフィールドとデータ型を示します。

汎用メタデータ

このメタデータは、すべての種類のストリームに共通するものです。

フィールド	Avro 型	JSON 型	説明
`stream_name`	文字列	文字列	作成時に定義された一意のストリーム名。
`read_method`	文字列	文字列	ソースからのデータの読み取りが、変更データキャプチャ（CDC）メソッドを使用したものか、履歴バックフィルの一部としてのものか、または CDC レプリケーション中にトランザクションがロールバックされたときに作成された補完タスクの一部としてのものかを示します。有効な値は次のとおりです。 `cdc` `backfill` `oracle-cdc-logminer` `oracle-backfill` `oracle-supplementation` `mysql-cdc-binlog` `mysql-backfill-incremental` `mysql-backfill-fulldump` `postgres-cdc-wal` `postgresql-backfill` `salesforce-cdc` `salesforce-backfill`
`object`	文字列	文字列	さまざまな種類のイベントをグループ化するために使用される名前。通常は、ソース内のテーブルかオブジェクトの名前です。
`schema_key`	文字列	文字列	イベントの統合スキーマの固有識別子。
`uuid`	文字列	文字列	Datastream によって生成されたイベントの固有識別子。
`read_timestamp`	timestamp-millis	文字列	Datastream がレコードを読み取ったときのタイムスタンプ（UTC）（ミリ秒単位のエポックタイムスタンプ）。
`source_timestamp`	timestamp-millis	文字列	ソースでレコードが変更されたときのタイムスタンプ（UTC）。ミリ秒単位のエポックタイムスタンプ。
`sort_keys`	`{"type": "array", "items": ["string", "long"]}`	配列	イベントを発生順に並べ替えるために使用できる値の配列。

ソース固有のメタデータ

このメタデータは、ソースデータベースの CDC イベントとバックフィルイベントに関連付けられています。このメタデータを表示するには、次のプルダウンメニューからソースを選択します。

ソース	フィールド	Avro 型	JSON 型	説明
MySQL	`log_file`	文字列	文字列	binlog ベースのレプリケーションを使用する場合に、Datastream が CDC レプリケーションでイベントを pull するログファイル。
MySQL	`log_position`	long	long	binlog ベースのレプリケーションを使用する場合の MySQL バイナリログのログ位置（オフセット）。
MySQL	`binlog_file`	文字列	文字列	GTID ベースのレプリケーションを使用する場合に、Datastream が CDC レプリケーションでイベントを pull するバイナリログファイル。
MySQL	`binlog_position`	long	long	GTID ベースのレプリケーションを使用する場合の MySQL バイナリログ内の位置（オフセット）。
MySQL	`gtid`	文字列	文字列	GTID ベースのレプリケーションを使用する場合に、イベントのトランザクションを一意に識別するグローバルトランザクション識別子（GTID）。
MySQL	`datastream_master_server_uuid`	文字列	文字列	GTID ベースのレプリケーションを使用するときに、Datastream がイベントを読み取る MySQL ソースサーバーインスタンスの UUID。
MySQL	`datastream_master_server_id`	long	long	GTID ベースのレプリケーションを使用するときに、Datastream がイベントを読み取る MySQL ソースサーバーインスタンスのサーバー ID。
MySQL	`primary_keys`	文字列配列	文字列配列	テーブルの主キーを構成する 1 つ以上の列名のリスト。テーブルに主キーがない場合、このフィールドは空です。
MySQL	`is_deleted`	ブール値	ブール値	`true` の値は、ソースで行が削除されたことを示します。 `false` の値は、行が削除されなかったことを示します。
MySQL	`database`	文字列	文字列	イベントに関連付けられているデータベース。
MySQL	`table`	文字列	文字列	イベントに関連付けられているテーブル。
MySQL	`change_type`	文字列	文字列	イベントが表す変更の種類（`INSERT`、`UPDATE-INSERT`、`UPDATE-DELETE`、`DELETE`）。 MySQL の行ベースのレプリケーションの場合、主キーの変更を含む行の各イベントには、更新する行を検索するときに列と照合される「前」のイメージと、変更を含む「後」のイメージの 2 つが含まれます。たとえば、`UPDATE-DELETE` の変更では、`UPDATE-DELETE` が「前」のイメージで、`UPDATE-INSERT` が「後」のイメージです。詳細は、MySQL の行ベースのレプリケーションをご覧ください。
Oracle	`log_file`	文字列	文字列	Datastream が CDC レプリケーションでイベントを pull するログファイル。
Oracle	`scn`	long	long	Oracle トランザクションログ内のログの位置（オフセット）。
Oracle	`row_id`	文字列	文字列	Oracle の row_id。
Oracle	`is_deleted`	ブール値	ブール値	`true` の値は、ソースで行が削除されたことを示します。 `false` の値は、行が削除されなかったことを示します。
Oracle	`database`	文字列	文字列	イベントに関連付けられているデータベース。
Oracle	`schema`	文字列	文字列	イベントからテーブルに関連付けられているスキーマ。
Oracle	`table`	文字列	文字列	イベントに関連付けられているテーブル。
Oracle	`change_type`	文字列	文字列	イベントが表す変更の種類（`INSERT`、`UPDATE-INSERT`、`UPDATE-DELETE`、`DELETE`）。 Oracle の行ベースのレプリケーションの場合、主キーの変更を含む行の各イベントには、更新する行を検索するときに列と照合される「前」のイメージと、変更を含む「後」のイメージの 2 つが含まれます。たとえば、`UPDATE-DELETE` の変更では、`UPDATE-DELETE` が「前」のイメージで、`UPDATE-INSERT` が「後」のイメージです。詳細は、Oracle の行ベースのレプリケーションをご覧ください。
Oracle	`tx_id`	文字列	文字列	イベントが属するトランザクション ID。
Oracle	`rs_id`	文字列	文字列	レコードセット ID。`rs_id` と `ssn` の組み合わせにより、`V$LOGMNR_CONTENTS` の行が一意に識別されます。`rs_id` は、行を生成した REDO レコードを一意に識別します。
Oracle	`ssn`	long	long	SQL シーケンス番号。この番号は `rs_id` とともに使用され、`V$LOGMNR_CONTENTS` の行を一意に識別します。
PostgreSQL	`schema`	文字列	文字列	イベントからテーブルに関連付けられているスキーマ。
PostgreSQL	`table`	文字列	文字列	イベントに関連付けられているテーブル。
PostgreSQL	`is_deleted`	ブール値	ブール値	`true` の値は、ソースで行が削除されたことを示します。 `false` の値は、行が削除されなかったことを示します。
PostgreSQL	`change_type`	文字列	文字列	イベントが表す変更の種類（`INSERT`、`UPDATE`、`DELETE`）。 PostgreSQL の行ベースのレプリケーションの場合、主キーの変更を含む行の各イベントには、更新する行を検索するときに列と照合される「前」のイメージと、変更を含む「後」のイメージの 2 つが含まれます。たとえば、`UPDATE-DELETE` の変更では、`UPDATE-DELETE` が「前」のイメージで、`UPDATE-INSERT` が「後」のイメージです。詳細は、PostgreSQL の行ベースのレプリケーションをご覧ください。
PostgreSQL	`tx_id`	文字列	文字列	イベントが属するトランザクション ID。
PostgreSQL	`lsn`	文字列	文字列	現在のエントリのログシーケンス番号。
PostgreSQL	`primary_keys`	文字列配列	文字列配列	テーブルの主キーを構成する 1 つ以上の列名のリスト。テーブルに主キーがない場合、このフィールドは空です。
SQL Server	`table`	文字列	文字列	イベントに関連付けられているテーブル。
SQL Server	`database`	long	long	イベントに関連付けられているデータベース。
SQL Server	`schema`	文字列配列	文字列配列	イベントからテーブルに関連付けられているスキーマ。
SQL Server	`is_deleted`	ブール値	ブール値	`true` の値は、ソースで行が削除されたことを示します。 `false` の値は、行が削除されなかったことを示します。
SQL Server	`lsn`	文字列	文字列	イベントのログシーケンス番号。
SQL Server	`tx_id`	文字列	文字列	イベントが属するトランザクション ID。
SQL Server	`physical_location`	整数配列	整数配列	3 つの整数（レコードのファイル ID、ページ ID、スロット ID）で記述されるログレコードの物理的な場所。
SQL Server	`replication_index`	文字列配列	文字列配列	テーブル内の行を一意に識別できるインデックスの列名のリスト。
SQL Server	`change_type`	文字列	string	イベントが表す変更の種類（`INSERT`、`UPDATE`、`DELETE`）。
Spanner	`commit_timestamp`	文字列	文字列	イベントの commit タイムスタンプ。
Spanner	`snapshot`	ブール値	ブール値	イベントがバックフィルスナップショットイベントかどうか。
Spanner	`project_id`	文字列	文字列	Spanner プロジェクト ID。
Spanner	`instance_id`	文字列	文字列	Spanner インスタンス ID。
Spanner	`database_id`	文字列	文字列	Spanner データベース識別子。
Spanner	`change_stream_name`	文字列	文字列	Spanner 変更ストリーム。
Spanner	`table`	文字列	文字列	Spanner テーブル。
Spanner	`server_transaction_id`	文字列	文字列	イベントに関連付けられたトランザクション ID。
Spanner	`record_sequence`	文字列	文字列	イベントに関連付けられたレコードシーケンス。
Spanner	`mod_index`	文字列	文字列	イベントに関連付けられた変更番号。
Spanner	`transaction_tag`	文字列	文字列	関連付けられたトランザクションのタグ。
Spanner	`system_transaction`	文字列	文字列	トランザクションがシステムトランザクションかどうか。
Spanner	`number_of_records_in_transaction`	long	long	関連するトランザクション内のレコード数。
Spanner	`value_capture_type`	enum	enum	変更ストリームの値キャプチャタイプ。
Spanner	`mod_type`	enum	enum	レコードが `INSERT`、`UPDATE`、`DELETE` のいずれか。
Spanner	`primary_keys`	配列	配列	テーブルの主キー列のリスト。
Spanner	`is_deleted`	ブール値	ブール値	変更が `DELETE` イベントを表すかどうか。
Salesforce	`object_name`	文字列	文字列	イベントに関連付けられた Salesforce オブジェクトの名前。
Salesforce	`domain`	文字列	文字列	イベントに関連付けられたドメインの名前。
Salesforce	`is_deleted`	ブール値	ブール値	`true` の値は、ソースで行が削除されたことを示します。 `false` の値は、行が削除されなかったことを示します。
Salesforce	`change_type`	文字列	string	イベントが表す変更の種類（`INSERT`、`UPDATE`、`DELETE`）。
Salesforce	`primary_keys`	文字列配列	文字列配列	テーブルの主キーを構成する列名のリスト。テーブルに主キーがない場合、このフィールドは空です。
MongoDB	`database`	文字列	文字列	イベントに関連付けられているデータベース。
MongoDB	`collection`	文字列	文字列	イベントに関連付けられているコレクション。コレクションは、リレーショナルデータベースのテーブルに似ています。
MongoDB	`change_type`	文字列	文字列	イベントが表す変更の種類（`CREATE`、`UPDATE`、`DELETE`）。
MongoDB	`is_deleted`	ブール値	ブール値	`true` の値は、ソースで行が削除されたことを示します。 `false` の値は、行が削除されなかったことを示します。
MongoDB	`primary_keys`	文字列配列	文字列配列	`_id` フィールド。コレクション内の各ドキュメントの主キーとして機能します。

イベントフローの例

このフローでは、ソースデータベースの SAMPLE テーブルの単一行に対して、3 つの連続したオペレーション（INSERT、UPDATE、DELETE）によって生成されるイベントを示します。

TIME	THIS_IS_MY_PK（int）	FIELD1（nchar、Null も可）	FIELD2（nchar、Null も可）>
0	1231535353	foo	TLV
1	1231535353	NULL	TLV

INSERT（T0）

メッセージペイロードは、新しい行全体で構成されます。

{
  "stream_name": "projects/myProj/locations/myLoc/streams/Oracle-to-Source",
  "read_method": "oracle-cdc-logminer",
  "object": "SAMPLE.TBL",
  "uuid": "d7989206-380f-0e81-8056-240501101100",
  "read_timestamp": "2019-11-07T07:37:16.808Z",
  "source_timestamp": "2019-11-07T02:15:39",  
  "source_metadata": {
    "log_file": ""
    "scn": 15869116216871,
    "row_id": "AAAPwRAALAAMzMBABD",
    "is_deleted": false,
    "database": "DB1",
    "schema": "ROOT",
    "table": "SAMPLE"
    "change_type": "INSERT",
    "tx_id": 
    "rs_id": "0x0073c9.000a4e4c.01d0",
    "ssn": 67,
  },
  "payload": {
    "THIS_IS_MY_PK": "1231535353",
    "FIELD1": "foo",
    "FIELD2": "TLV",
  }
}

UPDATE（T1）

メッセージペイロードは、新しい行全体で構成されます。以前の値は含まれません。

{
  "stream_name": "projects/myProj/locations/myLoc/streams/Oracle-to-Source",
  "read_method": "oracle-cdc-logminer",
  "object": "SAMPLE.TBL",
  "uuid": "e6067366-1efc-0a10-a084-0d8701101101",
  "read_timestamp": "2019-11-07T07:37:18.808Z",
  "source_timestamp": "2019-11-07T02:17:39",  
  "source_metadata": {
    "log_file": 
    "scn": 15869150473224,
    "row_id": "AAAGYPAATAAPIC5AAB",
    "is_deleted": false,
    "database":
    "schema": "ROOT",
    "table": "SAMPLE"
    "change_type": "UPDATE",
    "tx_id":
    "rs_id": "0x006cf4.00056b26.0010",
    "ssn": 0,
  },
  "payload": {
    "THIS_IS_MY_PK": "1231535353",
    "FIELD1": null,
    "FIELD2": "TLV",
  }
}

DELETE（T2）

メッセージペイロードは、新しい行全体で構成されます。

{
  "stream_name": "projects/myProj/locations/myLoc/streams/Oracle-to-Source",
  "read_method": "oracle-cdc-logminer",
  "object": "SAMPLE.TBL",
  "uuid": "c504f4bc-0ffc-4a1a-84df-6aba382fa651",
  "read_timestamp": "2019-11-07T07:37:20.808Z",
  "source_timestamp": "2019-11-07T02:19:39",
  "source_metadata": {
    "log_file": 
    "scn": 158691504732555,
    "row_id": "AAAGYPAATAAPIC5AAC",
    "is_deleted": true,
    "database":
    "schema": "ROOT",
    "table": "SAMPLE"
    "change_type": "DELETE",
    "tx_id":
    "rs_id": "0x006cf4.00056b26.0011",
    "ssn": 0,
  },
  "payload": {
    "THIS_IS_MY_PK": "1231535353",
    "FIELD1": null,
    "FIELD2": "TLV",
  }
}

順序と整合性

このセクションでは、Datastream が順序付けと整合性をどのように処理するかについて説明します。

注文

Datastream では、順序は保証されませんが、各イベントにはデータの行全体と、データがソースに書き込まれたときのタイムスタンプが含まれます。BigQuery では、順序が不適切なイベントは自動的に正しい順序にマージされます。BigQuery は、イベントメタデータと内部変更シーケンス番号（CSN）を使用して、イベントを正しい順序でテーブルに適用します。Cloud Storage では、同じ時刻のイベントが複数のファイルにまたがることがあります。

設計上、ストリームの開始時に作成されたデータの最初のバックフィルのためにイベントがバックフィルされた場合に、順不同で生成されるイベントが発生します。

順序はソースごとに推測できます。

ソース	説明
MySQL	イベントが初期バックフィルの一部の場合、`read_method` フィールドは `mysql-backfill` で始まります（バイナリログベースのレプリケーションの場合）。または `backfill` に設定されます（GTID ベースのレプリケーションの場合）。イベントは任意の順序で使用できるため、バックフィル内でイベントを受信する順序には影響しません。イベントが進行中のレプリケーションの一部の場合、`read_method` フィールドは `mysql-cdc-binlog`（binlog ベースのレプリケーションの場合）または `cdc`（GTID ベースのレプリケーションの場合）に設定されます。 binlog ベースのレプリケーションの場合、順序は `log_file` フィールドと `log_position` フィールドの組み合わせによって推測できます。GTID ベースのレプリケーションの場合、順序は `binlog_file` フィールドと `binlog_position` フィールド（または `gtid` フィールド）の組み合わせによって推測できます。この組み合わせにより、データベース内のオペレーションの順序を識別する一意の数値が徐々に増加します。
Oracle	イベントが最初のバックフィルの一部の場合、`read_method` フィールドは `oracle-backfill` で始まります。イベントは任意の順序で使用できるため、バックフィル内でイベントを受信する順序には影響しません。イベントが進行中のレプリケーションの一部の場合、`read_method` フィールドは `oracle-cdc-logminer` に設定されます。順序は `rs_id`（レコードセット ID）フィールドと `ssn`（SQL シーケンス番号）フィールドの組み合わせによって推測できます。この組み合わせにより、データベース内のオペレーションの順序を識別する一意の数値が徐々に増加します。
PostgreSQL	イベントが最初のバックフィルの一部の場合、`read_method` フィールドは `postgresql-backfill` で始まります。イベントは任意の順序で使用できるため、バックフィル内でイベントを受信する順序には影響しません。イベントが進行中のレプリケーションの一部の場合、`read_method` フィールドは `postgres-cdc-wal` に設定されます。順序は `source_timestamp` フィールドと `lsn`（ログシーケンス番号）フィールドの組み合わせによって推測できます。この組み合わせにより、データベース内のオペレーションの順序を識別する一意の数値が徐々に増加します。
SQL Server	イベントが最初のバックフィルの一部の場合、`read_method` フィールドは `sqlserver-backfill` で始まります。イベントは任意の順序で使用できるため、バックフィル内でイベントを受信する順序には影響しません。イベントが進行中のレプリケーションの一部の場合、`read_method` フィールドは `sqlserver-cdc` に設定されます。順序は `source_timestamp` フィールドと `lsn`（ログシーケンス番号）フィールドの組み合わせによって推測できます。この組み合わせにより、データベース内のオペレーションの順序を識別する一意の数値が徐々に増加します。 BigQuery にレプリケートする場合、データベースの作成時間も正しい順序を確保するために使用されます。これは、移行中やレプリカへのフェイルオーバー中など、ソースデータベースインスタンスが変更される可能性があるシナリオで重要です。
Salesforce	順序は、レコードの `source_timestamp` をソートキーとして使用することで決定できます。Salesforce のタイムスタンプの解像度は 1 秒ですが、同じレコードに対して同じ秒に 2 つの変更イベントが発生することはありません。
MongoDB	順序は、オペレーションログの `ts` フィールドまたはレコードの変更ストリームの `clusterTime` フィールドを使用して決定できます。フィールドはレコードごとに一意です。
Spanner	順序は、`commit_timestamp`、`record_sequence`、`mod_index` フィールドの組み合わせによって決定できます。この組み合わせにより、データベース内のオペレーションの順序を識別する一意の数値が徐々に増加します。

整合性

Datastream は、ソースデータベースからのデータが宛先に少なくとも 1 回配信されるようにします。イベントが失われることはありませんが、ストリーム内でイベントが重複する可能性があります。重複イベントのウィンドウは数分単位にする必要があります。イベントメタデータのイベントの汎用固有識別子（UUID）を使用して、重複を検出できます。

ただし、ダウンストリームでイベントの重複除去を行うために UUID のみに依存しないでください。パイプラインが中断された場合、Datastream は最後に保存されたチェックポイントからソースログを再読み取りして、配信を確実にします。Datastream はイベントの読み取り時に UUID を動的に生成するため、このタイプのリトライでは、まったく同じソースデータに対して新しい UUID が生成されます。また、ソースデータベースでの一括オペレーション中に、複数の異なる行が同じ変更シーケンス番号（CSN）を共有することがあります。

データの重複を確実に排除するには、ソーステーブルの主キーでイベントをパーティション分割し、変更シーケンス番号（CSN）（change_sequence_number フィールド）を使用してグループ化します。これにより、同じイベントの再試行が正しく折りたたまれるとともに、一括トランザクションから正当に区別された行の更新が保持されます。

データベースログファイルに commit されていないトランザクションが含まれている場合、トランザクションがロールバックされると、データベースはこれをログファイルに「逆」のデータ操作言語（DML）オペレーションとして反映します。たとえば、ロールバックした INSERT オペレーションには、対応する DELETE オペレーションが含まれます。Datastream は、これらのオペレーションをログファイルから読み取ります。

ストリームについて

すべてのストリームには、ストリームと、ストリームがデータを取得するソースの両方を記述するメタデータがあります。このメタデータには、ストリーム名、ソースと宛先の接続プロファイルなどの情報が含まれます。

Stream オブジェクトの完全な定義を確認するには、API リファレンスドキュメントをご覧ください。

ストリームの状態とステータス

ストリームがとり得る状態は、次のとおりです。

Not started
Starting
Running
Draining
Paused
Failed
Failed permanently

ログを使用して、テーブルのバックフィルや処理された行数などの追加のステータス情報を確認できます。FetchStreamErrors API を使用してエラーを取得することもできます。

Discover API で使用可能なオブジェクトメタデータ

discover API は、接続プロファイルで表されるデータソースまたは宛先で定義されたオブジェクトの構造を表すオブジェクトを返します。各オブジェクトには、オブジェクト自体のメタデータと、取得するデータの各フィールドのメタデータがあります。このメタデータは、discover API で取得できます。

次のステップ

ストリームの詳細については、ストリームのライフサイクルをご覧ください。
ストリームの作成方法については、ストリームを作成するをご覧ください。