MaxCompute:ストリーム

CDC システム列

read_mode を cdc に設定すると、出力行の末尾に 3 つのシステム列が追加されます。

UPDATE は DELETE／INSERT のペアとして表現されるため、__meta_op_type と __meta_is_update を組み合わせることで、正確な変更タイプを特定できます。

例

Delta テーブルを作成し、バージョン 1 から開始する append モードのストリームを作成します。

CREATE TABLE delta_table_src (
  pk bigint NOT NULL PRIMARY KEY,
  val bigint
) tblproperties ("transactional"="true");

CREATE STREAM delta_table_stream
ON TABLE delta_table_src VERSION AS OF 1
strmproperties('read_mode'='append')
comment 'Stream demo';

例

CREATE TABLE delta_table_src (pk BIGINT NOT NULL PRIMARY KEY,
val BIGINT) TBLPROPERTIES ("transactional"="true");

CREATE STREAM delta_table_stream ON TABLE delta_table_src
VERSION AS OF 1 strmproperties('read_mode'='append')
comment 'Stream demo';

DESC STREAM delta_table_stream;

出力例：

Name                                    delta_table_stream
Project                                 sql_optimizer
Create Time                             2024-09-06 17:03:32
Last Modified Time                      2024-09-06 17:03:32
Offset Version                          1
Reference Table Project                 sql_optimizer
Reference Table Name                    delta_table_src
Reference Table Id                      5e19a67eb97b4477b7fbce0c7bbcebca
Reference Table Version                 1
Parameters                              {
    "comment": "stream demo",
    "read_mode": "append"}

空のテーブルに対してストリームを初めて作成した場合、Offset Version と Reference Table Version は等しくなります。Delta テーブルに対して DML 操作が実行されると、Reference Table Version が進捗します。ストリームは (Offset Version, Reference Table Version] 内のすべての変更を返します。DML を用いた読み取りによりこれらの変更が消費されると、Offset Version が Reference Table Version に追いつき、新しい変更が到着するまでストリームは空の結果を返します。

ストリームプロパティの変更

ALTER STREAM <stream_name> SET strmproperties ("key"="value");

現在、ストリーム作成後に read_mode を変更することはできません。

初期データバージョンの変更

このコマンドを使用してオフセットバージョンをリセットします。たとえば、過去の変更範囲をスキップし、開始ポイントを進めることができます。

ALTER STREAM <stream_name> ON TABLE <delta_table_name>
<TIMESTAMP AS OF t | VERSION AS OF v>;

例

この例では、ストリームのライフサイクル全体を示します：ストリームの作成、Delta テーブルのバージョンを進めるためのデータ挿入、そしてストリームのオフセットバージョンのリセットです。

-- 1. ソース Delta テーブルを作成します。
CREATE TABLE delta_table_src (pk bigint NOT NULL PRIMARY KEY,
val bigint) tblproperties ("transactional"="true");

-- 2. バージョン 1 から開始するストリームを作成します。
CREATE STREAM delta_table_stream ON TABLE delta_table_src
VERSION AS OF 1 strmproperties('read_mode'='append')
comment 'Stream demo';

-- 3. オフセットバージョンと参照テーブルバージョンがともに 1 であることを確認します。
DESC STREAM delta_table_stream;
-- 出力例：
-- Offset Version          1
-- Reference Table Version 1

-- 4. Delta テーブルのバージョンを新しいものに進めるためにレコードを挿入します。
INSERT INTO delta_table_src VALUES ('1', '1');

-- 5. Delta テーブルのバージョン履歴を表示します。
SHOW HISTORY FOR TABLE delta_table_src;
-- ObjectType  ObjectId                          ObjectName       VERSION(LSN)      Time                  Operation
-- TABLE       8605276ce0034b20af761bf4761ba62e  delta_table_src  0000000000000001  2024-09-07 10:25:59   CREATE
-- TABLE       8605276ce0034b20af761bf4761ba62e  delta_table_src  0000000000000002  2024-09-07 10:28:19   APPEND

-- 6. ストリームのオフセットバージョンをバージョン 2 にリセットし、
--    ステップ 4 で挿入されたデータをスキップします。
ALTER STREAM delta_table_stream ON TABLE delta_table_src VERSION AS OF 2;

-- 7. 両方のバージョンが現在 2 であることを確認します。
DESC STREAM delta_table_stream;
-- 出力例：
-- Offset Version          2
-- Reference Table Version 2

例

-- 現在のプロジェクト内のすべてのストリームを一覧表示します。
SHOW STREAMS;
-- 出力例：
-- delta_table_stream

例

-- 1. ストリームが存在することを確認します。
SHOW STREAMS;
-- 出力例：
-- delta_table_stream

-- 2. ストリームを削除します。
DROP STREAM IF EXISTS delta_table_stream;

-- 3. ストリームが削除されたことを確認します。
SHOW STREAMS;
-- 出力例：（空）

例：CDC モード

この例では、ソーステーブルの INSERT および UPDATE を追跡し、CDC モードを使用して変更を送信先テーブルにコピーします。

Delta テーブルにおける CDC モードは招待プレビュー機能です。設定方法の詳細については、「CDC（招待プレビュー）」をご参照ください。

1. CDC を有効化したソース Delta テーブルを作成します。

   CREATE TABLE delta_table_src (
     pk bigint NOT NULL PRIMARY KEY,
     val bigint
   ) tblproperties (
     "transactional"="true",
     'acid.cdc.mode.enable'='true',
     'cdc.insert.into.passthrough.enable'='true'
   );

2. 送信先テーブルを作成します。

   CREATE TABLE delta_table_dest (
     pk bigint NOT NULL PRIMARY KEY,
     val bigint
   ) tblproperties ("transactional"="true");

3. CDC モードのストリームを作成します。

   CREATE STREAM delta_table_stream
   ON TABLE delta_table_src VERSION AS OF 1
   strmproperties('read_mode'='cdc')
   comment 'Stream cdc mode';

4. ソーステーブルに 2 件のレコードを挿入します。

   INSERT INTO delta_table_src VALUES (1, 1), (2, 2);

5. ストリームをクエリします。SELECT 文を単独で実行してもオフセットバージョンは進捗しないため、以降の実行でも同じ結果が返されます。

   SELECT * FROM delta_table_stream;
   
   -- 出力例
   +------------+------------+------------------+----------------+------------------+
   | pk         | val        | __meta_timestamp | __meta_op_type | __meta_is_update |
   +------------+------------+------------------+----------------+------------------+
   | 2          | 2          | 2024-09-07 11:03:53 | 1             | 0                |
   | 1          | 1          | 2024-09-07 11:03:53 | 1              | 0                |
   +------------+------------+------------------+----------------+------------------+

   両方の行で __meta_op_type=1（INSERT）および __meta_is_update=0（FALSE）が表示されており、これは新規挿入であることを示しています。

6. 送信先テーブルへの挿入により変更を消費します。これによりオフセットバージョンが進捗します。

   INSERT INTO delta_table_dest SELECT pk, val FROM delta_table_stream;

7. 送信先テーブルにデータが正しく受信されたことを確認します。

   SELECT * FROM delta_table_dest;
   
   -- 出力例
   +------------+------------+
   | pk         | val        |
   +------------+------------+
   | 1          | 1          |
   | 2          | 2          |
   +------------+------------+

8. 再度ストリームをクエリします。ステップ 6 で変更が消費されたため、空の結果が返されます。

   SELECT * FROM delta_table_stream;
   
   -- 出力例
   +------------+------------+
   | pk         | val        |
   +------------+------------+
   +------------+------------+

9. ソーステーブルの pk=1 を更新します。

   UPDATE delta_table_src SET val = 10 WHERE pk = 1;

10. 再度ストリームをクエリします。UPDATE は、更新前の状態と更新後の状態という 2 行で表示されます。

     SELECT * FROM delta_table_stream;
    
     -- 出力例
     +------------+------------+------------------+----------------+------------------+
     | pk         | val        | __meta_timestamp | __meta_op_type | __meta_is_update |
     +------------+------------+------------------+----------------+------------------+
     | 1          | 1          | 2024-09-07 11:10:21 | 0              | 1                |
     | 1          | 10         | 2024-09-07 11:10:21 | 1              | 1                |
     +------------+------------+------------------+----------------+------------------+

    1 行目（__meta_op_type=0、__meta_is_update=1）は更新前の値（DELETE + TRUE = UPDATE_BEFORE）であり、2 行目（__meta_op_type=1、__meta_is_update=1）は更新後の値（INSERT + TRUE = UPDATE_AFTER）です。

例：Append モード

この例では、UPDATE および DELETE 操作に対する Append モードと CDC モードの動作の違いを示します。

1. ソース Delta テーブルを作成します。

   CREATE TABLE delta_table_src (
     pk bigint NOT NULL PRIMARY KEY,
     val bigint
   ) tblproperties ("transactional"="true");

2. 送信先テーブルを作成します。

   CREATE TABLE delta_table_dest (
     pk bigint NOT NULL PRIMARY KEY,
     val bigint
   ) tblproperties ("transactional"="true");

3. Append モードのストリームを作成します。

   CREATE STREAM delta_table_stream
   ON TABLE delta_table_src VERSION AS OF 1
   strmproperties ('read_mode'='append')
   comment 'Stream append mode';

4. ソーステーブルに 2 件のレコードを挿入します。

   INSERT INTO delta_table_src VALUES (1, 1), (2, 2);

5. ストリームをクエリします。Append モードではシステム列は返されません。

   SELECT * FROM delta_table_stream;
   
   -- 出力：
   +------------+------------+
   | pk         | val        |
   +------------+------------+
   +------------+------------+

6. 変更内容を処理する。

   INSERT INTO delta_table_dest SELECT pk, val FROM delta_table_stream;

7. 送信先テーブルにデータが正しく受信されたことを確認します。

   SELECT * FROM delta_table_dest;
   
   -- 出力例
   +------------+------------+
   | pk         | val        |
   +------------+------------+
   | 1          | 1          |
   | 2          | 2          |
   +------------+------------+

8. ストリームをクエリします。ステップ 6 の変更が消費されたため、空の結果が返されます。

   SELECT * FROM delta_table_stream;
   
   -- 出力例
   +------------+------------+
   | pk         | val        |
   +------------+------------+
   +------------+------------+

9. ソーステーブルで pk=1 を更新し、pk=2 を削除します。

   UPDATE delta_table_src SET val = 10 WHERE pk = 1;
   DELETE FROM delta_table_src WHERE pk = 2;

10. ストリームをクエリします。

     SELECT * FROM delta_table_stream;
    
     -- 出力例
     +------------+------------+
     | pk         | val        |
     +------------+------------+
     | 1          | 10         |
     +------------+------------+

    更新された行 (1, 10) のみが返されます。削除された行は含まれません。Append モードは、変更された行の最終状態のみを返し、変更前イメージや削除操作を公開しません。継続的に挿入または更新されるデータを処理する ETL パイプラインには Append モードを、削除や更新前の値を含む完全な変更履歴を必要とする送信先システムには CDC モードを使用してください。