Incrementele belasting in Query-service

Maak een checkpoint_log -tabel waarin de meest recente opname wordt bijgehouden die is gebruikt om gegevens te verwerken. De volgende lijst (checkpoint_log in dit voorbeeld) moet eerst aan null worden geïnitialiseerd om een dataset stapsgewijs te verwerken.

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;

Vul de checkpoint_log lijst met één leeg verslag voor de dataset die stijgende verwerking vereist. DIM_TABLE_ABC is de gegevensset die in het onderstaande voorbeeld moet worden verwerkt. Bij de eerste keer dat DIM_TABLE_ABC wordt verwerkt, wordt last_snapshot_id geïnitialiseerd als null . Dit staat u toe om de volledige dataset bij de eerste gelegenheid en incrementeel daarna te verwerken.

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

Start vervolgens DIM_TABLE_ABC_Incremental om verwerkte uitvoer van DIM_TABLE_ABC te bevatten. Het anonieme blok in de vereiste uitvoeringssectie van het SQL hieronder voorbeeld, zoals die in stap één tot vier wordt beschreven, wordt opeenvolgend uitgevoerd om gegevens incrementeel te verwerken.

1. Stel de from_snapshot_id in die aangeeft vanaf welk punt de verwerking begint. De from_snapshot_id in het voorbeeld wordt vanuit de checkpoint_log -tabel opgevraagd voor gebruik met DIM_TABLE_ABC . Bij de eerste uitvoering zal de opname-id null zijn, wat betekent dat de volledige gegevensset wordt verwerkt.
2. Stel de to_snapshot_id in als de huidige opname-id van de brontabel (DIM_TABLE_ABC ). In het voorbeeld wordt dit gevraagd vanuit de metagegevenstabel van de brontabel.
3. Gebruik het trefwoord CREATE om DIM_TABLE_ABC_Incremenal te maken als de doeltabel. De bestemmingslijst voortduurt de verwerkte gegevens van de brondataset (DIM_TABLE_ABC). Hierdoor kunnen de verwerkte gegevens van de brontabel tussen from_snapshot_id en to_snapshot_id incrementeel aan de doeltabel worden toegevoegd.
4. Werk de checkpoint_log tabel bij met de to_snapshot_id voor de brongegevens die DIM_TABLE_ABC correct heeft verwerkt.
5. Als om het even welke opeenvolgend uitgevoerde vragen van het anonieme blok ontbreken, wordt de facultatieve uitzonderingssectie uitgevoerd. Dit retourneert een fout en beëindigt het proces.

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

Gebruik de stijgende logica van de gegevenslading in het anonieme blokvoorbeeld hieronder om het even welke nieuwe gegevens van de brondataset (sinds meest recente timestamp) toe te staan om aan de bestemmingstabel bij een regelmatige kadentie worden verwerkt en worden toegevoegd. In het voorbeeld worden gegevenswijzigingen in DIM_TABLE_ABC verwerkt en toegevoegd aan 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
$$;