Chargement incrémentiel dans Query Service

Créez une table checkpoint_log pour effectuer le suivi de l’instantané le plus récent utilisé pour traiter les données avec succès. La table de suivi (checkpoint_log dans cet exemple) doit d’abord être initialisé en null afin de traiter de manière incrémentielle un jeu de données.

DROP TABLE IF EXISTS checkpoint_log;
CREATE TABLE  checkpoint_log AS
SELECT
   cast(NULL AS string) process_name,
   cast(NULL AS string) process_status,
   cast(NULL AS string) last_snapshot_id,
   cast(NULL AS TIMESTAMP) process_timestamp
   WHERE false;

Renseignez la table checkpoint_log avec un enregistrement vide pour le jeu de données qui nécessite un traitement incrémentiel. DIM_TABLE_ABC est le jeu de données à traiter dans l’exemple ci-dessous. Lors du premier traitement de DIM_TABLE_ABC, l’last_snapshot_id est initialisé en tant que null. Cela vous permet de traiter l’ensemble du jeu de données la première fois et de manière incrémentielle par la suite.

INSERT INTO
   checkpoint_log
   SELECT
      'DIM_TABLE_ABC' process_name,
      'SUCCESSFUL' process_status,
      cast(NULL AS string) last_snapshot_id,
      CURRENT_TIMESTAMP process_timestamp;

Ensuite, initialisez DIM_TABLE_ABC_Incremental pour contenir la sortie traitée de DIM_TABLE_ABC. Le bloc anonyme dans la section d’exécution obligatoire de l’exemple SQL ci-dessous, comme décrit dans les étapes 1 à 4, est exécuté de manière séquentielle pour traiter les données de manière incrémentielle.

1. Définissez l’from_snapshot_id qui indique l’endroit où commence le traitement. L’from_snapshot_id dans l’exemple est interrogé à partir de la table checkpoint_log à utiliser avec DIM_TABLE_ABC. Lors de l’exécution initiale, l’ID de l’instantané est null, ce qui signifie que l’ensemble du jeu de données sera traité.
2. Définissez l’to_snapshot_id comme ID d’instantané actuel de la table source (DIM_TABLE_ABC). Dans l’exemple, cette requête provient de la table des métadonnées de la table source.
3. Utilisez le mot-clé CREATE pour créer DIM_TABLE_ABC_Incremenal comme table de destination. La table de destination conserve les données traitées du jeu de données source (DIM_TABLE_ABC). Cela permet aux données traitées de la table source entre from_snapshot_id et to_snapshot_id, d’être ajoutées de manière incrémentielle à la table de destination.
4. Mettez à jour la table checkpoint_log avec l’to_snapshot_id des données source que DIM_TABLE_ABC a traité avec succès.
5. Si l’une des requêtes exécutées de manière séquentielle du bloc anonyme échoue, la section d’exception facultative est exécutée. Cela renvoie une erreur et met fin au processus.

$$ BEGIN
    SET @from_snapshot_id = SELECT coalesce(last_snapshot_id, 'HEAD') FROM checkpoint_log a JOIN
                            (SELECT MAX(process_timestamp)process_timestamp FROM checkpoint_log
                                WHERE process_name = 'DIM_TABLE_ABC' AND process_status = 'SUCCESSFUL' )b
                                ON a.process_timestamp=b.process_timestamp;
    SET @to_snapshot_id = SELECT snapshot_id FROM (SELECT history_meta('DIM_TABLE_ABC')) WHERE  is_current = true;
    SET @last_updated_timestamp= SELECT CURRENT_TIMESTAMP;
    CREATE TABLE DIM_TABLE_ABC_Incremental AS
     SELECT  *  FROM DIM_TABLE_ABC SNAPSHOT BETWEEN @from_snapshot_id AND @to_snapshot_id ;

INSERT INTO
   checkpoint_log
   SELECT
       'DIM_TABLE_ABC' process_name,
       'SUCCESSFUL' process_status,
      cast( @to_snapshot_id AS string) last_snapshot_id,
      cast( @last_updated_timestamp AS TIMESTAMP) process_timestamp;

EXCEPTION
  WHEN OTHER THEN
    SELECT 'ERROR';
END
$$;

Utilisez la logique de chargement incrémentiel des données dans l’exemple de bloc anonyme ci-dessous pour permettre le traitement et l’ajout régulier de toutes les nouvelles données du jeu de données source (depuis la date et l’heure les plus récentes) à la table de destination. Dans l’exemple, les données modifiées en DIM_TABLE_ABC seront traitées et ajoutées à DIM_TABLE_ABC_incremental.

$$ BEGIN
    SET @from_snapshot_id = SELECT coalesce(last_snapshot_id, 'HEAD') FROM checkpoint_log a join
                            (SELECT MAX(process_timestamp)process_timestamp FROM checkpoint_log
                                WHERE process_name = 'DIM_TABLE_ABC' AND process_status = 'SUCCESSFUL' )b
                                ON a.process_timestamp=b.process_timestamp;
    SET @to_snapshot_id = SELECT snapshot_id FROM (SELECT history_meta('DIM_TABLE_ABC')) WHERE  is_current = true;
    SET @last_updated_timestamp= SELECT CURRENT_TIMESTAMP;
    INSERT INTO DIM_TABLE_ABC_Incremental
     SELECT  *  FROM DIM_TABLE_ABC SNAPSHOT BETWEEN @from_snapshot_id AND @to_snapshot_id WHERE NOT EXISTS (SELECT _id FROM DIM_TABLE_ABC_Incremental a WHERE _id=a._id);

INSERT INTO
   checkpoint_log
   SELECT
       'DIM_TABLE_ABC' process_name,
       'SUCCESSFUL' process_status,
      cast( @to_snapshot_id AS string) last_snapshot_id,
      cast( @last_updated_timestamp AS TIMESTAMP) process_timestamp;

EXCEPTION
  WHEN OTHER THEN
    SELECT 'ERROR';
END
$$;