Carga incremental en el servicio de consultas

Cree una tabla checkpoint_log para realizar el seguimiento de la instantánea más reciente utilizada para procesar datos correctamente. La tabla de seguimiento (checkpoint_log en este ejemplo) debe inicializarse primero en null para procesar de forma incremental un conjunto de datos.

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;

Rellene la tabla checkpoint_log con un registro vacío para el conjunto de datos que necesita procesamiento incremental. DIM_TABLE_ABC es el conjunto de datos que se procesará en el ejemplo siguiente. En la primera vez que se procese DIM_TABLE_ABC, last_snapshot_id se inicializó como null. Esto le permite procesar todo el conjunto de datos la primera vez e incrementalmente a partir de entonces.

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

A continuación, inicialice DIM_TABLE_ABC_Incremental para que contenga el resultado procesado de DIM_TABLE_ABC. El bloque anónimo de la sección de ejecución required del ejemplo de SQL siguiente, tal como se describe en los pasos del uno al cuatro, se ejecuta secuencialmente para procesar los datos de forma incremental.

1. Establezca from_snapshot_id, que indica desde dónde comienza el procesamiento. Se ha consultado el elemento from_snapshot_id del ejemplo en la tabla checkpoint_log para su uso con DIM_TABLE_ABC. En la primera ejecución, el ID de instantánea será null, lo que significa que se procesará todo el conjunto de datos.
2. Establezca to_snapshot_id como el identificador de instantánea actual de la tabla de origen (DIM_TABLE_ABC). En el ejemplo, esto se consulta desde la tabla de metadatos de la tabla de origen.
3. Use la palabra clave CREATE para crear DIM_TABLE_ABC_Incremenal como tabla de destino. La tabla de destino conserva los datos procesados del conjunto de datos de origen (DIM_TABLE_ABC). Esto permite que los datos procesados de la tabla de origen entre from_snapshot_id y to_snapshot_id se anexen gradualmente a la tabla de destino.
4. Actualice la tabla checkpoint_log con to_snapshot_id para los datos de origen que DIM_TABLE_ABC procesó correctamente.
5. Si falla cualquiera de las consultas ejecutadas secuencialmente del bloque anónimo, se ejecuta la sección de excepciones optional. Esto devuelve un error y finaliza el proceso.

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

Utilice la lógica de carga de datos incremental en el ejemplo de bloque anónimo siguiente para permitir que cualquier dato nuevo del conjunto de datos de origen (desde la marca de tiempo más reciente) se procese y se anexe a la tabla de destino a una cadencia regular. En el ejemplo, los cambios de datos en DIM_TABLE_ABC se procesarán y se anexarán 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
$$;