MaxCompute:Objets stream

CREATE STREAM [IF NOT EXISTS] <stream_name>
ON TABLE <delta_table_name> <TIMESTAMP AS OF t | VERSION AS OF v>
strmproperties ("read_mode"="append" | "cdc")
[comment <stream_comment>];

Colonnes système CDC

Lorsque read_mode est défini sur cdc, trois colonnes système sont ajoutées à chaque ligne de sortie :

Étant donné que les mises à jour sont représentées par une paire DELETE/INSERT, la combinaison de __meta_op_type et de __meta_is_update permet d'identifier le type exact de modification :

Exemple

Créez une table Delta, puis créez un stream en mode append à partir de la version 1.

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 <stream_name>;

Exemple

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;

Sortie :

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"}

Lors de la création initiale du stream sur une table vide, Offset Version et Reference Table Version sont identiques. À mesure que des opérations DML s'exécutent sur la table Delta, Reference Table Version avance. Le stream renvoie toutes les modifications situées dans (Offset Version, Reference Table Version] . Après qu'une lecture basée sur DML a consommé ces modifications, Offset Version rattrape Reference Table Version , et le stream renvoie un résultat vide jusqu'à l'arrivée de nouvelles modifications.

Modifier les propriétés d'un stream

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

Actuellement, read_mode ne peut pas être modifié après la création du stream.

Modifier la version de données initiale

Utilisez cette commande pour réinitialiser l'Offset Version, par exemple pour ignorer une plage de modifications historiques et faire avancer le point de départ.

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

Exemple

Cet exemple illustre le cycle de vie complet : création d'un stream, insertion de données pour faire avancer la version de la table Delta, puis réinitialisation de l'Offset Version du stream.

-- 1. Create a source Delta Table.
CREATE TABLE delta_table_src (pk bigint NOT NULL PRIMARY KEY,
val bigint) tblproperties ("transactional"="true");

-- 2. Create a stream starting from version 1.
CREATE STREAM delta_table_stream ON TABLE delta_table_src
VERSION AS OF 1 strmproperties('read_mode'='append')
comment 'Stream demo';

-- 3. Confirm that Offset Version and Reference Table Version are both 1.
DESC STREAM delta_table_stream;
-- Output:
-- Offset Version          1
-- Reference Table Version 1

-- 4. Insert a record to advance the Delta Table to a new version.
INSERT INTO delta_table_src VALUES ('1', '1');

-- 5. View Delta Table version history.
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. Reset the stream's Offset Version to version 2,
--    skipping the data inserted in step 4.
ALTER STREAM delta_table_stream ON TABLE delta_table_src VERSION AS OF 2;

-- 7. Confirm that both versions are now 2.
DESC STREAM delta_table_stream;
-- Output:
-- Offset Version          2
-- Reference Table Version 2

SHOW STREAMS;

Exemple

-- List all streams in the current project.
SHOW STREAMS;
-- Output:
-- delta_table_stream

DROP STREAM [IF EXISTS] <stream_name>;

Exemple

-- 1. Confirm the stream exists.
SHOW STREAMS;
-- Output:
-- delta_table_stream

-- 2. Delete the stream.
DROP STREAM IF EXISTS delta_table_stream;

-- 3. Confirm the stream is gone.
SHOW STREAMS;
-- Output: (empty)

SELECT * FROM <stream_name>;

INSERT INTO <destination_table> SELECT * FROM <stream_name>;

Exemple : mode CDC

Cet exemple suit les insertions et les mises à jour sur une table source et copie les modifications vers une table de destination en utilisant le mode CDC.

Le mode CDC sur les tables Delta nécessite une préversion sur invitation. Pour plus de détails sur la configuration, consultez CDC (préversion sur invitation).

1. Créez une table Delta source avec le CDC activé.

   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. Créez une table de destination.

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

3. Créez un stream en mode CDC.

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

4. Insérez deux enregistrements dans la table source.

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

5. Interrogez le stream. L'exécution de SELECT seule ne fait pas avancer l'Offset Version ; le même résultat est renvoyé à chaque exécution suivante.

   SELECT * FROM delta_table_stream;
   
   -- Output
   +------------+------------+------------------+----------------+------------------+
   | 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                |
   +------------+------------+------------------+----------------+------------------+

   Les deux lignes affichent __meta_op_type=1 (INSERT) et __meta_is_update=0 (FALSE), indiquant de nouvelles insertions.

6. Consommez les modifications en les insérant dans la table de destination. Cela fait avancer l'Offset Version.

   INSERT INTO delta_table_dest SELECT pk, val FROM delta_table_stream;

7. Confirmez que la table de destination a reçu les données.

   SELECT * FROM delta_table_dest;
   
   -- Output
   +------------+------------+
   | pk         | val        |
   +------------+------------+
   | 1          | 1          |
   | 2          | 2          |
   +------------+------------+

8. Interrogez à nouveau le stream. Il renvoie un résultat vide car les modifications ont été consommées à l'étape 6.

   SELECT * FROM delta_table_stream;
   
   -- Output
   +------------+------------+
   | pk         | val        |
   +------------+------------+
   +------------+------------+

9. Mettez à jour pk=1 dans la table source.

   UPDATE delta_table_src SET val = 10 WHERE pk = 1;

10. Interrogez à nouveau le stream. La mise à jour apparaît sous forme de deux lignes : l'état avant la mise à jour et l'état après la mise à jour.

     SELECT * FROM delta_table_stream;
    
     -- Output
     +------------+------------+------------------+----------------+------------------+
     | 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                |
     +------------+------------+------------------+----------------+------------------+

    La première ligne (__meta_op_type=0, __meta_is_update=1) correspond à la valeur avant la mise à jour (DELETE + TRUE = UPDATE_BEFORE). La deuxième ligne (__meta_op_type=1, __meta_is_update=1) correspond à la valeur après la mise à jour (INSERT + TRUE = UPDATE_AFTER).

Exemple : mode Append

Cet exemple montre la différence de comportement entre le mode append et le mode CDC pour les opérations de mise à jour et de suppression.

1. Créez une table Delta source.

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

2. Créez une table de destination.

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

3. Créez un stream en mode append.

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

4. Insérez deux enregistrements dans la table source.

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

5. Interrogez le stream. Le mode append ne renvoie aucune colonne système.

   SELECT * FROM delta_table_stream;
   
   -- Output
   +------------+------------+
   | pk         | val        |
   +------------+------------+
   | 1          | 1          |
   | 2          | 2          |
   +------------+------------+

6. Consommez les modifications.

   INSERT INTO delta_table_dest SELECT pk, val FROM delta_table_stream;

7. Confirmez que la table de destination a reçu les données.

   SELECT * FROM delta_table_dest;
   
   -- Output
   +------------+------------+
   | pk         | val        |
   +------------+------------+
   | 1          | 1          |
   | 2          | 2          |
   +------------+------------+

8. Interrogez le stream. Il renvoie un résultat vide — les modifications de l'étape 6 ont été consommées.

   SELECT * FROM delta_table_stream;
   
   -- Output
   +------------+------------+
   | pk         | val        |
   +------------+------------+
   +------------+------------+

9. Mettez à jour pk=1 et supprimez pk=2 dans la table source.

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

10. Interrogez le stream.

     SELECT * FROM delta_table_stream;
    
     -- Output
     +------------+------------+
     | pk         | val        |
     +------------+------------+
     | 1          | 10         |
     +------------+------------+

    Seule la ligne mise à jour (1, 10) est renvoyée. La ligne supprimée n'est pas incluse. Le mode append renvoie uniquement l'état final des lignes modifiées ; il n'expose ni les images avant modification ni les suppressions. Utilisez le mode append pour les pipelines ETL qui traitent des données continuellement insérées ou mises à jour ; utilisez le mode CDC lorsque votre système en aval a besoin de l'historique complet des modifications, y compris les suppressions et les valeurs avant mise à jour.