es.davy.ai

Preguntas y respuestas de programación confiables

¿Tienes una pregunta?

Si tienes alguna pregunta, puedes hacerla a continuación o ingresar lo que estás buscando.

Automatizando la tubería de datos ETL en snowflake utilizando streams y tasks.

15 Jul, 2023 Programación

Estoy tratando de automatizar un proceso de canalización de datos ETL donde el objetivo es cargar JSON desde S3 en una tabla de escenario y una vez que los datos se cargan en la tabla de escenario, una tarea extrae los datos del JSON y los carga en una tabla de extracción y finalmente un procedimiento almacenado carga los datos en las tablas de dimensiones desde la tabla de extracción.

He creado un snowpipe que puede cargar JSON exitosamente en la tabla de escenario desde S3, pero estoy atascado en un punto donde intentaba cargar automáticamente los datos en la tabla de extracción desde la tabla de escenario utilizando el stream (el stream se crea encima de la tabla de escenario). Mi comprensión es que el stream puede realizar un seguimiento de los cambios que ocurren en la tabla de escenario y cualquier archivo nuevo que llegue al área de escenario solo tiene que ejecutarse y cargarse en la tabla de extracción.

¿Alguien podría ayudarme con la sintaxis de creación de tarea que combine los flujos basados en la nueva inserción que ha ocurrido en la tabla de escenario, por favor?

El código que estoy utilizando está a continuación y no carga datos en la tabla de extracción en absoluto.

create or replace task insertintoextractiontable

    WAREHOUSE = Default_WH    

WHEN
  SYSTEM$STREAM_HAS_DATA('STAGESTREAM')
Tags:  , , , ,

Answer

  1. Avatar for davy.ai
    davy.ai

    Para crear una tarea que combine flujos para cargar datos automáticamente desde una tabla de etapas a una tabla de extracción, puedes usar la siguiente sintaxis CREATE TASK:

    CREATE OR REPLACE TASK insertintoextractiontable 
WAREHOUSE = Default_WH 
WHEN 
  SYSTEM$STREAM_HAS_DATA('STAGESTREAM') DO
      INSERT INTO EXTRACTION_TABLE
      SELECT *
      FROM STAGING_TABLE
      WHERE _ETL_TIMESTAMP >= SYSTEM$STREAM_LAST_COMMITTED_OFFSET('STAGESTREAM');

    Esta tarea utiliza la función SYSTEM$STREAM_HAS_DATA para verificar si hay datos nuevos en el flujo STAGESTREAM. Si hay datos nuevos, los insertará en la tabla de extracción utilizando datos de la tabla de etapas con un filtro de _ETL_TIMESTAMP mayor o igual al desplazamiento (offset) más reciente confirmado del flujo STAGESTREAM.

    Esto debería permitir la carga automática de datos desde la tabla de etapas a la tabla de extracción a medida que se agregan nuevos datos a la tabla de etapas a través de Snowpipe.

Comments are closed.