How to Set up a CICD Pipeline for Snowflake to Automate Data Pipelines

				
					-- V1__INVENTORY_TABLE_DDL.SQL
CRATE OR REPLACE TABLE STAGE.INVENTORY(
   ID INT,
   COLUMN1 NUMBER,
   COLUMN2 DATE,
   COLUMN3 VARCHAR
);

-- V2__FILE_FORMAT_DDL.SQL
CREARE OR REPLACE FILE FORMAT STAGE.CSV_FILE_FORMAT
   TYPE = 'CSV'
   FIELD_DELIMITER = ','
   SKIP_HEADER = 1;

-- V3__STAGE_DDL.SQL
CREATE STAGE STAGE.MY_S3_STAGE
 STORAGE_INTEGRATION = S3_INT
 URL = 'S3://MYBUCKET/ENCRYPTED_FILES/'
 FILE_FORMAT = STAGE.CSV_FILE_FORMAT;

-- V4__PIPE_DDL.SQL
CREATE OR REPLACE PIPE STAGE.MY_PIPE
   AUTO_INGEST = TRUE
   AS
COPY INTO STAGE.INVENTORY FROM @MY_S3_STAGE

-- V5__STREAM_DDL.SQL
CREATE OR REPLACE STREAM STAGE.INVENTORY_STM
   ON TABLE STAGE.INVENTORY
   APPEND_ONLY = TRUE;

-- V6__CURATED_TABLE_DDL.SQL
CRATE OR REPLACE TABLE CURATED.INVENTORY(
   ID INT,
   COLUMN1 NUMBER,
   COLUMN2 DATE,
   COLUMN3 VARCHAR
);

-- V7__TASK_DDL.SQL
CREATE OR REPLACE TASK STAGE.MERGE_INVETORY
   WAREHOUSE = ETL_WH
   SCHEDULE = '60 MINUTE'
   WHEN
   SYSTEM$STREAM_HAS_DATA('STAGE.INVENTORY_STM')
       AS
   MERGE INTO CURATED.INVENTORY TARGET
   USING STAGE.INVENTORY_STM SOURCE ON
     CURATED.INVENTORY.ID = STAGE.INVENTORY_STM.ID
   WHEN MATCHED
      THEN UPDATE SET 
       TARGET.COLUMN1 = SOURCE.COLUMN1,
       TARGET.COLUMN2 = SOURCE.COLUMN2,
       TARGET.COLUMN3 = SOURCE.COLUMN3
   WHEN NOT MATCHED THEN
    INSERT (ID,COLUMN1,COLUMN2,COLUMN3)
      VALUES(
        SOURCE.ID,
        SOURCE.COLUMN1,
        SOURCE.COLUMN2,
        SOURCE.COLUMN3
      );


ALTER TASK MERGE_INVETORY RESUME;