このページでは、変更ストリームの次の属性について詳しく説明します。
- スプリットベースのパーティショニング モデル
- 変更ストリーム レコードの形式と内容
- これらのレコードのクエリに使用される低レベルの構文
- クエリ ワークフローの例
このページの情報は、Spanner API を使用して変更ストリームを直接クエリする場合に最も役立ちます。そうではなく、Dataflow を使用して変更ストリーム データを読み取るアプリケーションの場合、ここで説明するデータモデルを直接操作する必要はありません。
変更ストリームの概要については、変更ストリームの概要をご覧ください。
変更ストリーム パーティション
変更ストリームが監視しているテーブルが変更されると、Spanner はデータ変更と同じトランザクションで同期的に、対応する変更ストリーム レコードをデータベースに書き込みます。これにより確実に、トランザクションが成功した場合に、Spanner は変更を正常にキャプチャし、保持することになります。内部的には、Spanner は変更ストリーム レコードとデータ変更を同じ場所に配置します。このようにして同じサーバーで処理されるようにすることで、書き込みのオーバーヘッドを最小限に抑えるようにします。
特定のスプリットに対する DML の一環として、Spanner は同じトランザクションで対応する変更ストリーム データ スプリットに書き込みを追加します。このコロケーションにより、変更ストリームは提供リソース間で余分な調整を追加せず、トランザクションの commit オーバーヘッドが最小限に抑えられます。
Spanner は、データベースの負荷とサイズに基づいてデータを動的に分割およびマージし、サービス リソース間でスプリットを分散させてスケーリングします。
変更ストリームの書き込みと読み取りのスケーリングを可能にするために、Spanner は内部の変更ストリーム ストレージをデータベース データとともに分割し、マージするため、自動的にホットスポットを防止します。データベースの書き込みのスケーリングに合わせて、変更ストリーム レコードをニア リアルタイムで読み取りできるようにするため、Spanner API は変更ストリーム パーティションを使用して同時にクエリが実行されるように設計されています。変更ストリーム パーティションは、変更ストリーム レコードを含む変更ストリーム データ スプリットにマッピングされます。変更ストリーム パーティションは時間の経過とともに動的に変更され、Spanner がデータベース データを動的に分割およびマージする方法と関連付けられます。
変更ストリームのパーティションには、特定の期間における不変のキー範囲のレコードが含まれています。変更ストリーム パーティションは、1 つ以上の変更ストリーム パーティションに分割することも、他の変更ストリーム パーティションと統合することもできます。これらの分割イベントまたは統合イベントが発生すると、子パーティションが作成され、次の期間のそれぞれの不変のキー範囲の変更がキャプチャされます。データ変更レコードに加えて、変更ストリーム クエリは、クエリが必要な新しい変更ストリーム パーティションのリーダーに通知する子パーティション レコードと、最近発生した書き込みがない場合の転送の進行状況を示すハートビート レコードを返します。
特定の変更ストリーム パーティションに対してクエリを実行すると、変更レコードは commit タイムスタンプ順に返されます。各変更レコードは 1 回だけ返されます。変更ストリーム パーティション間で、変更レコードの順序が保証されません。特定の主キーの変更レコードは、特定の期間の 1 つのパーティションでのみ返されます。
親子パーティションのリネージのため、commit タイムスタンプ順で特定のキーの変更を処理するには、子パーティションから返されたレコードは、すべての親パーティションからのレコードが処理された後にのみ処理する必要があります。
変更ストリームの読み取り機能とクエリ構文
GoogleSQL
変更ストリームにクエリを実行するには、ExecuteStreamingSql
API を使用します。Spanner は、変更ストリームとともに特別な読み取り関数を自動的に作成します。読み取り関数は、変更ストリームのレコードへのアクセス権を付与します。読み取り関数の命名規則は READ_change_stream_name
です。
データベースに変更ストリーム SingersNameStream
が存在すると仮定すると、GoogleSQL のクエリ構文は次のようになります。
SELECT ChangeRecord
FROM READ_SingersNameStream (
start_timestamp,
end_timestamp,
partition_token,
heartbeat_milliseconds,
read_options
)
読み取り関数は、次の引数を受け入れます。
引数名 | タイプ | 必須 | 説明 |
---|---|---|---|
start_timestamp |
TIMESTAMP |
必須 | commit_timestamp が start_timestamp 以上であるレコードを返すように指定します。値は変更ストリームの保持期間内にする必要があり、現在の時刻以下で、変更ストリームの作成のタイムスタンプ以上である必要があります。 |
end_timestamp |
TIMESTAMP |
省略可(デフォルト: NULL ) |
commit_timestamp が end_timestamp 以下であるレコードを返すように指定します。値は変更ストリームの保持期間内にする必要があり、start_timestamp 以上である必要があります。クエリは、end_timestamp までのすべての変更レコードを返した後、またはユーザーが接続を終了した後に終了します。NULL の場合、または指定されていない場合、すべての変更レコードが返されるか、ユーザーが接続を終了するまでクエリが実行されます。 |
partition_token |
STRING |
省略可(デフォルト: NULL ) |
子パーティション レコードの内容に基づいて、クエリする変更ストリーム パーティションを指定します。NULL または指定されていない場合、リーダーが初めて変更ストリームをクエリし、クエリに使用する特定のパーティション トークンを取得していないことを意味します。 |
heartbeat_milliseconds |
INT64 |
必須 | このパーティションに commit されたトランザクションがない場合に、ハートビートの ChangeRecord が返される頻度を決定します。
値は 1,000 (1 秒)から 300,000 (5 分)の範囲で指定してください。 |
read_options |
ARRAY |
省略可(デフォルト: NULL ) |
将来の使用のために予約された追加の読み取りオプション。現在、指定できる値は NULL のみです。 |
次の例に示すように、読み取り関数クエリのテキストを構築し、そのテキストにパラメータをバインドするための便利なメソッドを作成することをおすすめします。
Java
private static final String SINGERS_NAME_STREAM_QUERY_TEMPLATE = "SELECT ChangeRecord FROM READ_SingersNameStream" + "(" + " start_timestamp => @startTimestamp," + " end_timestamp => @endTimestamp," + " partition_token => @partitionToken," + " heartbeat_milliseconds => @heartbeatMillis" + ")"; // Helper method to conveniently create change stream query texts and bind parameters. public static Statement getChangeStreamQuery( String partitionToken, Timestamp startTimestamp, Timestamp endTimestamp, long heartbeatMillis) { return Statement.newBuilder(SINGERS_NAME_STREAM_QUERY_TEMPLATE) .bind("startTimestamp") .to(startTimestamp) .bind("endTimestamp") .to(endTimestamp) .bind("partitionToken") .to(partitionToken) .bind("heartbeatMillis") .to(heartbeatMillis) .build(); }
PostgreSQL
変更ストリームにクエリを実行するには、ExecuteStreamingSql
API を使用します。Spanner は、変更ストリームとともに特別な読み取り関数を自動的に作成します。読み取り関数は、変更ストリームのレコードへのアクセス権を付与します。読み取り関数の命名規則は spanner.read_json_change_stream_name
です。
データベースに変更ストリーム SingersNameStream
が存在すると仮定すると、PostgreSQL のクエリ構文は次のようになります。
SELECT *
FROM "spanner"."read_json_SingersNameStream" (
start_timestamp,
end_timestamp,
partition_token,
heartbeat_milliseconds,
null
)
読み取り関数は、次の引数を受け入れます。
引数名 | タイプ | 必須 | 説明 |
---|---|---|---|
start_timestamp |
timestamp with time zone |
必須 | commit_timestamp が start_timestamp 以上である変更レコードを返すように指定します。値は変更ストリームの保持期間内にする必要があり、現在の時刻以下で、変更ストリームの作成のタイムスタンプ以上である必要があります。 |
end_timestamp |
timestamp with timezone |
省略可(デフォルト: NULL ) |
commit_timestamp が end_timestamp 以下である変更レコードを返すように指定します。値は変更ストリームの保持期間内にする必要があり、start_timestamp 以上である必要があります。
クエリは、end_timestamp までのすべての変更レコードを返すか、ユーザーが接続を終了した後に終了します。
NULL の場合、すべての変更レコードが返されるか、ユーザーが接続を終了するまでクエリが実行されます。 |
partition_token |
text |
省略可(デフォルト: NULL ) |
子パーティション レコードの内容に基づいて、クエリする変更ストリーム パーティションを指定します。NULL または指定されていない場合、リーダーが初めて変更ストリームをクエリし、クエリに使用する特定のパーティション トークンを取得していないことを意味します。 |
heartbeat_milliseconds |
bigint |
必須 | このパーティションに commit されたトランザクションがない場合に、ハートビートの ChangeRecord が返される頻度を決定します。
値は 1,000 (1 秒)から 300,000 (5 分)の範囲で指定してください。 |
null |
null |
必須 | 将来の使用のために予約 |
次の例に示すように、読み取り関数のテキストを構築し、そのテキストにパラメータをバインドするための便利なメソッドを作成することをおすすめします。
Java
private static final String SINGERS_NAME_STREAM_QUERY_TEMPLATE = "SELECT * FROM \"spanner\".\"read_json_SingersNameStream\"" + "($1, $2, $3, $4, null)"; // Helper method to conveniently create change stream query texts and bind parameters. public static Statement getChangeStreamQuery( String partitionToken, Timestamp startTimestamp, Timestamp endTimestamp, long heartbeatMillis) { return Statement.newBuilder(SINGERS_NAME_STREAM_QUERY_TEMPLATE) .bind("p1") .to(startTimestamp) .bind("p2") .to(endTimestamp) .bind("p3") .to(partitionToken) .bind("p4") .to(heartbeatMillis) .build(); }
変更ストリームのレコード形式
GoogleSQL
変更ストリームの読み取り関数は、ARRAY<STRUCT<...>>
型の単一の ChangeRecord
列を返します。各行のこの配列には、常に 1 つの要素が含まれます。
配列要素の型は次のとおりです。
STRUCT <
data_change_record ARRAY<STRUCT<...>>,
heartbeat_record ARRAY<STRUCT<...>>,
child_partitions_record ARRAY<STRUCT<...>>
>
この構造体には、data_change_record
、heartbeat_record
、child_partitions_record
の 3 つのフィールドがあり、それぞれ ARRAY<STRUCT<...>>
型です。変更ストリーム読み取り関数が返す行では、これらの 3 つのフィールドの 1 つにのみ値が含まれ、他の 2 つは空または NULL
です。これらの配列フィールドには、最大で 1 つの要素が含まれます。
以降のセクションでは、これらの 3 つのレコードタイプについて説明します。
PostgreSQL
変更ストリームの読み取り関数は、次の構造を持つ JSON
型の単一の ChangeRecord
列を返します。
{
"data_change_record" : {},
"heartbeat_record" : {},
"child_partitions_record" : {}
}
このオブジェクトのキーには、data_change_record
、heartbeat_record
、child_partitions_record
の 3 つがあり、対応する値の型は JSON
です。変更ストリーム読み取り関数が返す行には、これらの 3 つのキーのいずれか 1 つだけが存在します。
以降のセクションでは、これらの 3 つのレコードタイプについて説明します。
データ変更レコード
データ変更レコードには、同じトランザクションの 1 つの変更ストリーム パーティション内の同じ commit タイムスタンプで commit された同じ変更タイプ(挿入、更新、削除)を持つテーブルに対する一連の変更が含まれます。複数のデータ変更レコードが、複数の変更ストリーム パーティション間の同じトランザクションで返されることがあります。
すべてのデータ変更レコードには commit_timestamp
、server_transaction_id
、record_sequence
の各フィールドがあり、これらを組み合わせてストリーム レコードの変更ストリーム内の順序が決まります。これらの 3 つのフィールドは、変更の順序を導き出し、外部整合性を提供するのに十分です。
重複しないデータにアクセスする複数のトランザクションが、同じ commit タイムスタンプを持つ場合があります。server_transaction_id
フィールドを使用すると、同じトランザクション内で発行された変更のセット(潜在的に変更ストリーム パーティション間など)を区別できます。record_sequence
フィールドおよび number_of_records_in_transaction
フィールドと組み合わせると、特定のトランザクションのすべてのレコードをバッファリングして順序付けすることもできます。
データ変更レコードのフィールドには、次のものが含まれます。
GoogleSQL
フィールド | タイプ | 説明 |
---|---|---|
commit_timestamp |
TIMESTAMP |
変更が commit されたときのタイムスタンプ |
record_sequence |
STRING |
トランザクション内のレコードのシーケンス番号。シーケンス番号は、一意で、単調に増加する(必ずしも連続ではない)ことが保証されています。同じ server_transaction_id のレコードを record_sequence で並べ替えて、トランザクション内の変更の順序を再構築します。この順序は、パフォーマンスを向上させるために Spanner によって最適化される場合があり、ユーザーが指定する元の順序と一致しないことがあります |
server_transaction_id |
STRING |
変更が commit された Spanner トランザクションを表す、グローバルに一意の文字列。この値は、変更ストリームのレコードを処理するコンテキストでのみ使用し、Spanner の API のトランザクション ID とは関連付けられません。 |
is_last_record_in_transaction_in_partition |
BOOL |
これが現在のパーティション内のトランザクションの最後のレコードかどうかを示します。 |
table_name |
STRING |
変更の影響を受けるテーブルの名前。 |
value_capture_type |
STRING |
この変更がキャプチャされたときに変更ストリーム構成で指定された値キャプチャ タイプを記述します。 値キャプチャ タイプは、 |
column_types |
ARRAY<STRUCT< |
列の名前、列のタイプ、主キーかどうか、スキーマで定義されている列の位置(「ordinal_position」など)。スキーマ内のテーブルの最初の列は、1 の順序位置になります。配列列の場合、列のタイプをネストできます。形式は、Spanner API リファレンスで説明されている型構造と一致します。 |
mods |
ARRAY<STRUCT< |
主キーの値、古い値、変更または追跡列の新しい値など、行われた変更について説明します。
古い値と新しい値の可用性と内容は、構成された value_capture_type によって異なります。new_values フィールドと old_values フィールドには、非キー列のみが含まれます。 |
mod_type |
STRING |
変更のタイプを表します。INSERT 、UPDATE 、DELETE のいずれかです。 |
number_of_records_in_transaction |
INT64 |
すべての変更ストリーム パーティションのこのトランザクションに含まれるデータ変更レコードの数。 |
number_of_partitions_in_transaction |
INT64 |
このトランザクションのデータ変更レコードを返すパーティションの数。 |
transaction_tag |
STRING |
このトランザクションに関連付けられたトランザクション タグ。 |
is_system_transaction |
BOOL |
トランザクションがシステム トランザクションかどうかを示します。 |
PostgreSQL
フィールド | タイプ | 説明 |
---|---|---|
commit_timestamp |
STRING |
変更が commit されたときのタイムスタンプ |
record_sequence |
STRING |
トランザクション内のレコードのシーケンス番号。シーケンス番号は、一意で、単調に増加する(必ずしも連続するではない)ことが保証されています。同じ `server_transaction_id` のレコードを `record_sequence` で並べ替えて、トランザクション内の変更の順序を再構築します。 |
server_transaction_id |
STRING |
変更が commit された Spanner トランザクションを表す、グローバルに一意の文字列。この値は、変更ストリームのレコードを処理するコンテキストでのみ使用し、Spanner の API のトランザクション ID とは関連付けられません。 |
is_last_record_in_transaction_in_partition |
BOOLEAN |
これが現在のパーティション内のトランザクションの最後のレコードかどうかを示します。 |
table_name |
STRING |
変更の影響を受けるテーブルの名前。 |
value_capture_type |
STRING |
この変更がキャプチャされたときに変更ストリーム構成で指定された値キャプチャ タイプを記述します。 値キャプチャ タイプは、 |
column_types |
[ { "name": <STRING>, "type": { "code": <STRING> }, "is_primary_key": <BOOLEAN>, "ordinal_position": <NUMBER> }, ... ] |
列の名前、列のタイプ、主キーかどうか、スキーマで定義されている列の位置(「ordinal_position」など)。スキーマ内のテーブルの最初の列は、1 の順序位置になります。配列列の場合、列のタイプをネストできます。形式は、Spanner API リファレンスで説明されている型構造と一致します。 |
mods |
[ { "keys": {<STRING> : <STRING>}, "new_values": { <STRING> : <VALUE-TYPE>, [...] }, "old_values": { <STRING> : <VALUE-TYPE>, [...] }, }, [...] ] |
主キーの値、古い値、変更または追跡列の新しい値など、行われた変更について説明します。古い値と新しい値の可用性と内容は、構成された value_capture_type によって異なります。new_values フィールドと old_values フィールドには、非キー列のみが含まれます。 |
mod_type |
STRING |
変更のタイプを表します。INSERT 、UPDATE 、DELETE のいずれかです。 |
number_of_records_in_transaction |
INT64 |
すべての変更ストリーム パーティションのこのトランザクションに含まれるデータ変更レコードの数。 |
number_of_partitions_in_transaction |
NUMBER |
このトランザクションのデータ変更レコードを返すパーティションの数。 |
transaction_tag |
STRING |
このトランザクションに関連付けられたトランザクション タグ。 |
is_system_transaction |
BOOLEAN |
トランザクションがシステム トランザクションかどうかを示します。 |
以下に、データ変更レコードの例を示します。2 つのアカウント間での転送がある単一のトランザクションを記述します。2 つのアカウントは個別の変更ストリーム パーティションにあります。
"data_change_record": {
"commit_timestamp": "2022-09-27T12:30:00.123456Z",
// record_sequence is unique and monotonically increasing within a
// transaction, across all partitions.
"record_sequence": "00000000",
"server_transaction_id": "6329047911",
"is_last_record_in_transaction_in_partition": true,
"table_name": "AccountBalance",
"column_types": [
{
"name": "AccountId",
"type": {"code": "STRING"},
"is_primary_key": true,
"ordinal_position": 1
},
{
"name": "LastUpdate",
"type": {"code": "TIMESTAMP"},
"is_primary_key": false,
"ordinal_position": 2
},
{
"name": "Balance",
"type": {"code": "INT"},
"is_primary_key": false,
"ordinal_position": 3
}
],
"mods": [
{
"keys": {"AccountId": "Id1"},
"new_values": {
"LastUpdate": "2022-09-27T12:30:00.123456Z",
"Balance": 1000
},
"old_values": {
"LastUpdate": "2022-09-26T11:28:00.189413Z",
"Balance": 1500
},
}
],
"mod_type": "UPDATE", // options are INSERT, UPDATE, DELETE
"value_capture_type": "OLD_AND_NEW_VALUES",
"number_of_records_in_transaction": 2,
"number_of_partitions_in_transaction": 2,
"transaction_tag": "app=banking,env=prod,action=update",
"is_system_transaction": false,
}
"data_change_record": {
"commit_timestamp": "2022-09-27T12:30:00.123456Z",
"record_sequence": "00000001",
"server_transaction_id": "6329047911",
"is_last_record_in_transaction_in_partition": true,
"table_name": "AccountBalance",
"column_types": [
{
"name": "AccountId",
"type": {"code": "STRING"},
"is_primary_key": true,
"ordinal_position": 1
},
{
"name": "LastUpdate",
"type": {"code": "TIMESTAMP"},
"is_primary_key": false,
"ordinal_position": 2
},
{
"name": "Balance",
"type": {"code": "INT"},
"is_primary_key": false,
"ordinal_position": 3
}
],
"mods": [
{
"keys": {"AccountId": "Id2"},
"new_values": {
"LastUpdate": "2022-09-27T12:30:00.123456Z",
"Balance": 2000
},
"old_values": {
"LastUpdate": "2022-01-20T11:25:00.199915Z",
"Balance": 1500
},
},
...
],
"mod_type": "UPDATE", // options are INSERT, UPDATE, DELETE
"value_capture_type": "OLD_AND_NEW_VALUES",
"number_of_records_in_transaction": 2,
"number_of_partitions_in_transaction": 2,
"transaction_tag": "app=banking,env=prod,action=update",
"is_system_transaction": false,
}
次のデータ変更レコードは、値のキャプチャ タイプが "NEW_VALUES"
のレコードの例です。新しい値のみが入力されることに注意してください。
変更されたのは "LastUpdate"
列のみであるため、その列のみが返されました。
"data_change_record": {
"commit_timestamp": "2022-09-27T12:30:00.123456Z",
// record_sequence is unique and monotonically increasing within a
// transaction, across all partitions.
"record_sequence": "00000000",
"server_transaction_id": "6329047911",
"is_last_record_in_transaction_in_partition": true,
"table_name": "AccountBalance",
"column_types": [
{
"name": "AccountId",
"type": {"code": "STRING"},
"is_primary_key": true,
"ordinal_position": 1
},
{
"name": "LastUpdate",
"type": {"code": "TIMESTAMP"},
"is_primary_key": false,
"ordinal_position": 2
}
],
"mods": [
{
"keys": {"AccountId": "Id1"},
"new_values": {
"LastUpdate": "2022-09-27T12:30:00.123456Z"
},
"old_values": {}
}
],
"mod_type": "UPDATE", // options are INSERT, UPDATE, DELETE
"value_capture_type": "NEW_VALUES",
"number_of_records_in_transaction": 1,
"number_of_partitions_in_transaction": 1,
"transaction_tag": "app=banking,env=prod,action=update",
"is_system_transaction": false
}
次のデータ変更レコードは、値のキャプチャ タイプが "NEW_ROW"
のレコードの例です。"LastUpdate"
列のみが変更されていますが、トラッキング対象のすべての列が返されます。
"data_change_record": {
"commit_timestamp": "2022-09-27T12:30:00.123456Z",
// record_sequence is unique and monotonically increasing within a
// transaction, across all partitions.
"record_sequence": "00000000",
"server_transaction_id": "6329047911",
"is_last_record_in_transaction_in_partition": true,
"table_name": "AccountBalance",
"column_types": [
{
"name": "AccountId",
"type": {"code": "STRING"},
"is_primary_key": true,
"ordinal_position": 1
},
{
"name": "LastUpdate",
"type": {"code": "TIMESTAMP"},
"is_primary_key": false,
"ordinal_position": 2
},
{
"name": "Balance",
"type": {"code": "INT"},
"is_primary_key": false,
"ordinal_position": 3
}
],
"mods": [
{
"keys": {"AccountId": "Id1"},
"new_values": {
"LastUpdate": "2022-09-27T12:30:00.123456Z",
"Balance": 1000
},
"old_values": {}
}
],
"mod_type": "UPDATE", // options are INSERT, UPDATE, DELETE
"value_capture_type": "NEW_ROW",
"number_of_records_in_transaction": 1,
"number_of_partitions_in_transaction": 1,
"transaction_tag": "app=banking,env=prod,action=update",
"is_system_transaction": false
}
次のデータ変更レコードは、値のキャプチャ タイプが "NEW_ROW_AND_OLD_VALUES"
のレコードの例です。"LastUpdate"
列のみが変更されていますが、トラッキング対象のすべての列が返されます。 この値キャプチャ タイプは、LastUpdate
の新しい値と古い値をキャプチャします。
"data_change_record": {
"commit_timestamp": "2022-09-27T12:30:00.123456Z",
// record_sequence is unique and monotonically increasing within a
// transaction, across all partitions.
"record_sequence": "00000000",
"server_transaction_id": "6329047911",
"is_last_record_in_transaction_in_partition": true,
"table_name": "AccountBalance",
"column_types": [
{
"name": "AccountId",
"type": {"code": "STRING"},
"is_primary_key": true,
"ordinal_position": 1
},
{
"name": "LastUpdate",
"type": {"code": "TIMESTAMP"},
"is_primary_key": false,
"ordinal_position": 2
},
{
"name": "Balance",
"type": {"code": "INT"},
"is_primary_key": false,
"ordinal_position": 3
}
],
"mods": [
{
"keys": {"AccountId": "Id1"},
"new_values": {
"LastUpdate": "2022-09-27T12:30:00.123456Z",
"Balance": 1000
},
"old_values": {
"LastUpdate": "2022-09-26T11:28:00.189413Z"
}
}
],
"mod_type": "UPDATE", // options are INSERT, UPDATE, DELETE
"value_capture_type": "NEW_ROW_AND_OLD_VALUES",
"number_of_records_in_transaction": 1,
"number_of_partitions_in_transaction": 1,
"transaction_tag": "app=banking,env=prod,action=update",
"is_system_transaction": false
}
ハートビート レコード
ハートビート レコードが返されると、ハートビート レコードの timestamp
以下の commit_timestamp
を持つすべての変更が返され、このパーティション内の将来のデータレコードには、ハートビート レコードによって返されるものよりも高い commit タイムスタンプが必要であることを示しています。パーティションにデータの変更が書き込まれていない場合、ハートビート レコードが返されます。パーティションにデータ変更が書き込まれた場合、heartbeat_record.timestamp
の代わりに data_change_record.commit_timestamp
を使用して、リーダーがパーティションの読み取りを進めていることを示すことができます。
パーティションで返されたハートビート レコードを使用して、すべてのパーティションでリーダーを同期できます。すべてのリーダーがあるタイムスタンプ A
以上のハートビートを受信するか、タイムスタンプ A
以上のデータまたは子パーティションのレコードを受信すると、リーダーは、そのタイムスタンプ A
以前に commit されたすべてのレコードを受信したことを認識し、バッファ内のレコードの処理を開始できます(たとえば、タイムスタンプでクロス パーティションのレコードを並べ替え、server_transaction_id
でグループ化)。
ハートビート レコードには、次の 1 つのフィールドのみが含まれます。
GoogleSQL
フィールド | タイプ | 説明 |
---|---|---|
timestamp |
TIMESTAMP |
ハートビート レコードのタイムスタンプ。 |
PostgreSQL
フィールド | タイプ | 説明 |
---|---|---|
timestamp |
STRING |
ハートビート レコードのタイムスタンプ。 |
このレコードのタイムスタンプと等しいか、それ以前のタイムスタンプを持つすべてのレコードが返されたことを示すハートビート レコードの例:
heartbeat_record: {
"timestamp": "2022-09-27T12:35:00.312486Z"
}
子パーティション レコード
子パーティション レコードは、子パーティションに関する情報(パーティション トークン、親パーティションのトークン、子パーティションに含まれる変更レコードの最も古いタイムスタンプを表す start_timestamp
)を返します。commit タイムスタンプが child_partitions_record.start_timestamp
の直前のレコードであるレコードが現在のパーティションに返されます。このパーティションの子パーティション レコードがすべて返されると、このクエリは成功ステータスを返します。これは、このパーティションのすべてのレコードが返されていることを示します。
子パーティション レコードのフィールドには、以下が含まれます。
GoogleSQL
フィールド | タイプ | 説明 |
---|---|---|
start_timestamp |
TIMESTAMP |
この子パーティション レコードの子パーティションから返されるデータ変更レコードは、start_timestamp 以上の commit タイムスタンプを持ちます。子パーティションをクエリする場合は、子パーティション トークンと、child_partitions_token.start_timestamp 以上の start_timestamp をクエリで指定する必要があります。パーティションによって返されるすべての子パーティション レコードは同じ start_timestamp を持ち、タイムスタンプは常にクエリで指定された start_timestamp と end_timestamp の間にあります。 |
record_sequence |
STRING |
単調に増加するシーケンス番号。特定のパーティションで同じ start_timestamp で返された子パーティション レコードが複数ある場合に、子パーティション レコードの順序を定義するために使用できます。パーティション トークン start_timestamp と record_sequence は、子パーティション レコードを一意に識別します。 |
child_partitions |
ARRAY<STRUCT< |
子パーティションのセットとそれに関連する情報を返します。これには、クエリで子パーティションを識別するために使用されるパーティション トークン文字列と、その親パーティションのトークンが含まれます。 |
PostgreSQL
フィールド | タイプ | 説明 |
---|---|---|
start_timestamp |
STRING |
この子パーティション レコードの子パーティションから返されるデータ変更レコードは、start_timestamp 以上の commit タイムスタンプを持ちます。子パーティションをクエリする場合は、子パーティション トークンと、child_partitions_token.start_timestamp 以上の start_timestamp をクエリで指定する必要があります。パーティションによって返されるすべての子パーティション レコードは同じ start_timestamp を持ち、タイムスタンプは常にクエリで指定された start_timestamp と end_timestamp の間にあります。
|
record_sequence |
STRING |
単調に増加するシーケンス番号。特定のパーティションで同じ start_timestamp で返された子パーティション レコードが複数ある場合に、子パーティション レコードの順序を定義するために使用できます。パーティション トークン start_timestamp と record_sequence は、子パーティション レコードを一意に識別します。 |
child_partitions |
[ { "token": <STRING>, "parent_partition_tokens": [<STRING>], }, [...] ] |
子パーティションの配列とそれに関連する情報を返します。これには、クエリで子パーティションを識別するために使用されるパーティション トークン文字列と、その親パーティションのトークンが含まれます。 |
子パーティション レコードの例を次に示します。
child_partitions_record: {
"start_timestamp": "2022-09-27T12:40:00.562986Z",
"record_sequence": "00000001",
"child_partitions": [
{
"token": "child_token_1",
// To make sure changes for a key is processed in timestamp
// order, wait until the records returned from all parents
// have been processed.
"parent_partition_tokens": ["parent_token_1", "parent_token_2"]
}
],
}
変更ストリームのクエリ ワークフロー
単回使用の読み取り専用トランザクションと強力な ExecuteStreamingSql
API と タイムスタンプ バウンドを使用して、変更ストリーム クエリを実行します。変更ストリーム読み取り関数を使用すると、対象の時間範囲の start_timestamp
と end_timestamp
を指定できます。保持期間内のすべての変更レコードには、強力な読み取り専用タイムスタンプ バウンドを使用してアクセスできます。
他のすべての TransactionOptions
は、変更ストリーム クエリでは無効です。さらに、TransactionOptions.read_only.return_read_timestamp
が true に設定されている場合、トランザクションを読み取る Transaction
メッセージで、有効なタイムスタンプの読み取りの代わりに、特別な値 kint64max - 1
が返されます。この特殊な値を破棄し、後続のクエリには使用しないでください。
各変更ストリーム クエリは、任意の数の行を返すことができ、各行にはデータ変更レコード、ハートビート レコード、子パーティション レコードのいずれかが含まれます。リクエストの期限を設定する必要はありません。
例:
ストリーミング クエリ ワークフローは、partition_token
を NULL
に指定して最初の変更ストリーム クエリを発行することから始まります。クエリでは、変更ストリームの読み取り関数、対象の開始タイムスタンプと終了タイムスタンプ、ハートビート間隔を指定する必要があります。end_timestamp
が NULL
の場合、クエリはパーティションが終了するまでデータ変更を返し続けます。
GoogleSQL
SELECT ChangeRecord FROM READ_SingersNameStream (
start_timestamp => "2022-05-01T09:00:00Z",
end_timestamp => NULL,
partition_token => NULL,
heartbeat_milliseconds => 10000
);
PostgreSQL
SELECT *
FROM "spanner"."read_json_SingersNameStream" (
'2022-05-01T09:00:00Z',
NULL,
NULL,
10000,
NULL
) ;
子パーティション レコードが返されるまで、このクエリからのデータレコードを処理します。次の例では、2 つの子パーティション レコードと 3 つのパーティション トークンが返され、クエリが終了します。特定のクエリからの子パーティション レコードは常に同じ start_timestamp
を共有します。
child_partitions_record: {
"record_type": "child_partitions",
"start_timestamp": "2022-05-01T09:00:01Z",
"record_sequence": 1000012389,
"child_partitions": [
{
"token": "child_token_1",
// Note parent tokens are null for child partitions returned
// from the initial change stream queries.
"parent_partition_tokens": [NULL]
}
{
"token": "child_token_2",
"parent_partition_tokens": [NULL]
}
],
}
child partitions record: {
"record_type": "child_partitions",
"start_timestamp": "2022-05-01T09:00:01Z",
"record_sequence": 1000012390,
"child_partitions": [
{
"token": "child_token_3",
"parent_partition_tokens": [NULL]
}
],
}
2022-05-01T09:00:01Z
以降の変更を処理するには、3 つの新しいクエリを作成し、並行して実行します。3 つのクエリを組み合わせると、親がカバーする同じキー範囲の将来のデータ変更が返されます。start_timestamp
は常に同じ子パーティション レコードの start_timestamp
に設定し、同じ end_timestamp
とハートビート間隔を使用して、すべてのクエリでレコードを一貫して処理します。
GoogleSQL
SELECT ChangeRecord FROM READ_SingersNameStream (
start_timestamp => "2022-05-01T09:00:01Z",
end_timestamp => NULL,
partition_token => "child_token_1",
heartbeat_milliseconds => 10000
);
SELECT ChangeRecord FROM READ_SingersNameStream (
start_timestamp => "2022-05-01T09:00:01Z",
end_timestamp => NULL,
partition_token => "child_token_2",
heartbeat_milliseconds => 10000
);
SELECT ChangeRecord FROM READ_SingersNameStream (
start_timestamp => "2022-05-01T09:00:01Z",
end_timestamp => NULL,
partition_token => "child_token_3",
heartbeat_milliseconds => 10000
);
PostgreSQL
SELECT *
FROM "spanner"."read_json_SingersNameStream" (
'2022-05-01T09:00:01Z',
NULL,
'child_token_1',
10000,
NULL
);
SELECT *
FROM "spanner"."read_json_SingersNameStream" (
'2022-05-01T09:00:01Z',
NULL,
'child_token_2',
10000,
NULL
);
SELECT *
FROM "spanner"."read_json_SingersNameStream" (
'2022-05-01T09:00:01Z',
NULL,
'child_token_3',
10000,
NULL
);
しばらくすると、別の子パーティション レコードが返されると child_token_2
のクエリが終了します。このレコードは、child_token_2
と child_token_3
の両方で、新しいパーティションが2022-05-01T09:30:15Z
カバーされます。まったく同じレコードが、child_token_3
に対するクエリで返されます。これは、両方とも新しい child_token_4
の親パーティションであるためです。
特定のキーのデータレコードの厳密な順序付けされた処理を保証するためには、child_token_4
に対するクエリはすべての親が完了してから開始する必要があります。この場合は、child_token_2
と child_token_3
です。子パーティション トークンごとに 1 つのクエリのみを作成します。クエリ ワークフローの設計では、child_token_4
でクエリを待機してスケジュールを設定する親を 1 つ指名する必要があります。
child partitions record: {
"record_type": "child_partitions",
"start_timestamp": "2022-05-01T09:30:15Z",
"record_sequence": 1000012389,
"child_partitions": [
{
"token": "child_token_4",
"parent_partition_tokens": [child_token_2, child_token_3],
}
],
}
GoogleSQL
SELECT ChangeRecord FROM READ_SingersNameStream(
start_timestamp => "2022-05-01T09:30:15Z",
end_timestamp => NULL,
partition_token => "child_token_4",
heartbeat_milliseconds => 10000
);
PostgreSQL
SELECT *
FROM "spanner"."read_json_SingersNameStream" (
'2022-05-01T09:30:15Z',
NULL,
'child_token_4',
10000,
NULL
);
Apache Beam SpannerIO Dataflow コネクタで変更ストリーム レコードを処理して解析する例については、GitHub をご覧ください。