Carico incrementale in Query Service

Creare una tabella checkpoint_log per tenere traccia dello snapshot più recente utilizzato per elaborare correttamente i dati. La tabella di rilevamento (checkpoint_log in questo esempio) deve prima essere inizializzata in null per elaborare in modo incrementale un set di dati.

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;

Popolare la tabella checkpoint_log con un record vuoto per il set di dati che richiede l'elaborazione incrementale. DIM_TABLE_ABC è il set di dati da elaborare nell'esempio seguente. Alla prima elaborazione di DIM_TABLE_ABC, last_snapshot_id è inizializzato come null. Questo consente di elaborare l’intero set di dati la prima volta e in modo incrementale in seguito.

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

Inizializzare quindi DIM_TABLE_ABC_Incremental per contenere l'output elaborato da DIM_TABLE_ABC. Il blocco anonimo nella sezione di esecuzione required dell'esempio SQL seguente, come descritto nei passaggi da uno a quattro, viene eseguito in sequenza per elaborare i dati in modo incrementale.

1. Imposta from_snapshot_id che indica da dove inizia l'elaborazione. Viene eseguita una query per from_snapshot_id nell'esempio dalla tabella checkpoint_log per l'utilizzo con DIM_TABLE_ABC. All'esecuzione iniziale, l'ID snapshot sarà null, il che significa che l'intero set di dati verrà elaborato.
2. Imposta to_snapshot_id come ID snapshot corrente della tabella di origine (DIM_TABLE_ABC). Nell’esempio, questa viene eseguita dalla tabella dei metadati della tabella di origine.
3. Utilizzare la parola chiave CREATE per creare DIM_TABLE_ABC_Incremenal come tabella di destinazione. La tabella di destinazione mantiene i dati elaborati dal set di dati di origine (DIM_TABLE_ABC). Ciò consente di aggiungere in modo incrementale alla tabella di destinazione i dati elaborati dalla tabella di origine tra from_snapshot_id e to_snapshot_id.
4. Aggiornare la tabella checkpoint_log con to_snapshot_id per i dati di origine elaborati correttamente da DIM_TABLE_ABC.
5. Se una delle query eseguite in sequenza del blocco anonimo non riesce, viene eseguita la sezione dell'eccezione optional. Questo restituisce un errore e termina il processo.

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

Utilizza la logica di caricamento dati incrementale nell’esempio di blocco anonimo seguente per consentire a tutti i nuovi dati dal set di dati di origine (dal timestamp più recente) di essere elaborati e aggiunti alla tabella di destinazione a una cadenza regolare. Nell'esempio, le modifiche dei dati a DIM_TABLE_ABC verranno elaborate e aggiunte a 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
$$;