Snowflake Streams - Working Session скачать в хорошем качестве

Snowflake Streams - Working Session

USE DATABASE MYOWN_DB; CREATE SCHEMA IF NOT EXISTS MYSTREAMS; CREATE SCHEMA IF NOT EXISTS STAGE_TBLS; CREATE SCHEMA IF NOT EXISTS INTG_TBLS; CREATE TABLE STAGE_TBLS.STG_EMPL ( EMPID INT, EMPNAME VARCHAR(30), SALARY FLOAT, AGE INT, DEPT VARCHAR(10), LOCATION VARCHAR(20) ); CREATE STREAM MYSTREAMS.STREAM_EMPL ON TABLE STAGE_TBLS.STG_EMPL; CREATE STREAM MYSTREAMS.STREAM_EMPL_2 ON TABLE STAGE_TBLS.STG_EMPL; SHOW STREAMS IN SCHEMA MYSTREAMS; SELECT * FROM MYSTREAMS.STREAM_EMPL; CREATE TABLE INTG_TBLS.EMPL ( EMPID INT,EMPNAME VARCHAR(30),SALARY FLOAT,AGE INT,DEPT VARCHAR(15), LOCATION VARCHAR(20),INSRT_DT DATE,LST_UPDT_DT DATE ); //Inserts INSERT INTO STAGE_TBLS.STG_EMPL VALUES (1, 'Amar', 80000, 35, 'SALES', 'Bangalore'), (2, 'Bharath', 45000, 26, 'SALES', 'Hyderabad'), (3, 'Charan', 76000, 34, 'TECHNOLOGY', 'Chennai'), (4, 'Divya', 52000, 28, 'HR', 'Hyderabad'), (5, 'Gopal', 24500, 22, 'TECHNOLOGY', 'Bangalore'), (6, 'Haritha', 42000, 27, 'HR', 'Chennai') ; SELECT * FROM STAGE_TBLS.STG_EMPL; SELECT * FROM MYSTREAMS.STREAM_EMPL; INSERT INTO INTG_TBLS.EMPL ( EMPID, EMPNAME, SALARY, AGE, DEPT, LOCATION, INSRT_DT, LST_UPDT_DT) SELECT EMPID, EMPNAME, SALARY, AGE, DEPT, LOCATION, CURRENT_DATE, NULL FROM MYSTREAMS.STREAM_EMPL WHERE METADATA$ACTION = 'INSERT' AND METADATA$ISUPDATE = FALSE; SELECT * FROM INTG_TBLS.EMPL; SELECT * FROM MYSTREAMS.STREAM_EMPL; //Updates SELECT * FROM STAGE_TBLS.STG_EMPL; UPDATE STAGE_TBLS.STG_EMPL SET SALARY=49000 WHERE EMPID=2; UPDATE STAGE_TBLS.STG_EMPL SET LOCATION='Pune' WHERE EMPID=5; SELECT * FROM STAGE_TBLS.STG_EMPL; SELECT * FROM MYSTREAMS.STREAM_EMPL; MERGE INTO INTG_TBLS.EMPL E USING MYSTREAMS.STREAM_EMPL S ON E.EMPID = S.EMPID WHEN MATCHED AND S.METADATA$ACTION ='INSERT' AND S.METADATA$ISUPDATE ='TRUE' THEN UPDATE SET E.EMPNAME = S.EMPNAME, E.SALARY = S.SALARY, E.AGE = S.AGE, E.DEPT = S.DEPT, E.LOCATION = S.LOCATION, E.LST_UPDT_DT = CURRENT_DATE; SELECT * FROM INTG_TBLS.EMPL; SELECT * FROM MYSTREAMS.STREAM_EMPL; //Deletes SELECT * FROM STAGE_TBLS.STG_EMPL; DELETE FROM STAGE_TBLS.STG_EMPL WHERE EMPID in (3,4); SELECT * FROM STAGE_TBLS.STG_EMPL; SELECT * FROM MYSTREAMS.STREAM_EMPL; MERGE INTO INTG_TBLS.EMPL E USING MYSTREAMS.STREAM_EMPL S ON E.EMPID = S.EMPID WHEN MATCHED AND S.METADATA$ACTION ='DELETE' AND S.METADATA$ISUPDATE ='FALSE' THEN DELETE; SELECT * FROM INTG_TBLS.EMPL; SELECT * FROM MYSTREAMS.STREAM_EMPL; //All changes at a time SELECT * FROM STAGE_TBLS.STG_EMPL; INSERT INTO STAGE_TBLS.STG_EMPL VALUES (7, 'Janaki', 61000, 29, 'SALES', 'Pune'), (8, 'Kamal', 92000, 33, 'TECHNOLOGY', 'Bangalore'); UPDATE STAGE_TBLS.STG_EMPL SET SALARY=85000, LOCATION='Hyderabad' WHERE EMPID=1; DELETE FROM STAGE_TBLS.STG_EMPL WHERE EMPID in (6); SELECT * FROM STAGE_TBLS.STG_EMPL; SELECT * FROM MYSTREAMS.STREAM_EMPL; MERGE INTO INTG_TBLS.EMPL T USING MYSTREAMS.STREAM_EMPL S ON T.EMPID = S.EMPID WHEN MATCHED AND S.METADATA$ACTION ='DELETE' AND S.METADATA$ISUPDATE = 'FALSE' THEN DELETE WHEN MATCHED AND S.METADATA$ACTION ='INSERT' AND S.METADATA$ISUPDATE = 'TRUE' THEN UPDATE SET T.EMPNAME = S.EMPNAME, T.SALARY = S.SALARY, T.AGE = S.AGE, T.DEPT = S.DEPT, T.LOCATION = S.LOCATION, T.LST_UPDT_DT = CURRENT_DATE WHEN NOT MATCHED AND S.METADATA$ACTION ='INSERT' AND S.METADATA$ISUPDATE = 'FALSE' THEN INSERT( EMPID, EMPNAME, SALARY, AGE, DEPT, LOCATION, INSRT_DT, LST_UPDT_DT) VALUES(S.EMPID, S.EMPNAME, S.SALARY, S.AGE, S.DEPT, S.LOCATION, CURRENT_DATE, NULL); SELECT * FROM INTG_TBLS.EMPL; SELECT * FROM MYSTREAMS.STREAM_EMPL; //With Tasks CREATE OR REPLACE TASK MYTASKS.TASK_EMPL_DATA_LOAD WAREHOUSE = MYOWN_WH SCHEDULE = '5 MINUTES' WHEN SYSTEM$STREAM_HAS_DATA('MYSTREAMS.STREAM_EMPL') AS MERGE INTO INTG_TBLS.EMPL T USING MYSTREAMS.STREAM_EMPL S ON T.EMPID = S.EMPID WHEN MATCHED AND S.METADATA$ACTION ='DELETE' AND S.METADATA$ISUPDATE = 'FALSE' THEN DELETE WHEN MATCHED AND S.METADATA$ACTION ='INSERT' AND S.METADATA$ISUPDATE = 'TRUE' THEN UPDATE SET T.EMPNAME = S.EMPNAME, T.SALARY = S.SALARY, T.AGE = S.AGE, T.DEPT = S.DEPT, T.LOCATION = S.LOCATION, T.LST_UPDT_DT = CURRENT_DATE WHEN NOT MATCHED AND S.METADATA$ACTION ='INSERT' AND S.METADATA$ISUPDATE = 'FALSE' THEN INSERT( EMPID, EMPNAME, SALARY, AGE, DEPT, LOCATION, INSRT_DT, LST_UPDT_DT) VALUES(S.EMPID, S.EMPNAME, S.SALARY, S.AGE, S.DEPT, S.LOCATION, CURRENT_DATE, NULL); ALTER TASK MYTASKS.TASK_EMPL_DATA_LOAD RESUME; SELECT * FROM STAGE_TBLS.STG_EMPL; INSERT INTO STAGE_TBLS.STG_EMPL VALUES (9, 'Latha', 47000, 25, 'HR', 'Chennai'); UPDATE STAGE_TBLS.STG_EMPL SET SALARY=67000 WHERE EMPID=7; DELETE FROM STAGE_TBLS.STG_EMPL WHERE EMPID in (8); SELECT * FROM MYSTREAMS.STREAM_EMPL; SELECT * FROM INTG_TBLS.EMPL; SELECT * FROM MYSTREAMS.STREAM_EMPL;