Automate the process using Task and Streams with snowflake
Learn how to enhance your workflow using Snowflake's powerful automation and streaming capabilities. This video provides a step-by-step guide to automating tasks, boosting productivity, and managing your data more efficiently with Snowflake. Ideal for anyone looking to optimize their data processes and save time!
#Snowflake #DataAutomation #WorkflowOptimization #DataStreaming #CloudData #AutomationTools #ProductivityBoost #DataManagement #StreamlineProcesses #TechTutorial #DataSolutions #EfficientData #CloudComputing #TechGuides #digitaltransformation
-----------------------------------------------------------------------------------------------------
// Stream example: INSERT ------------------------
CREATE OR REPLACE DATABASE STREAMS_DB;
-- Create customers table
CREATE OR REPLACE TABLE customers (
customer_id INT,
customer_name STRING,
email STRING,
phone STRING
);
-- Insert data into customers table
INSERT INTO customers (customer_id, customer_name, email, phone) VALUES
(101, 'Alice Smith', '[email protected]', '123-456-7890'),
(102, 'Bob Johnson', '[email protected]', '234-567-8901'),
(103, 'Charlie Brown', '[email protected]', '345-678-9012');
-- Create orders table
CREATE OR REPLACE TABLE orders_row_staging (
order_id INT,
customer_id INT,
order_date DATE,
status STRING,
total_amount NUMERIC(10, 2)
);
-- Insert data into orders table
INSERT INTO orders_row_staging (order_id, customer_id, order_date, status, total_amount) VALUES
(301, 101, '2024-05-01', 'Pending', 1499.97),
(302, 102, '2024-05-02', 'Shipped', 499.99),
(303, 103, '2024-05-03', 'Delivered', 299.99);
-- creating final table for storing data
create or replace table orders_final_table(
order_id INT,
order_date DATE,
status STRING,
total_amount NUMERIC(10, 2),
customer_name STRING,
email STRING,
phone STRING
);
-- Insert into final table
INSERT INTO orders_final_table
select order_id ,order_date ,status ,total_amount, customer_name, email, phone
from orders_row_staging a inner join customers b on a.customer_id = b.customer_id;
-- Create a stream object
create or replace stream orders_stream on table orders_row_staging;
SHOW STREAMS;
DESC STREAM orders_stream;
-- Get changes on data using stream (INSERTS)
select * from orders_stream;
select * from orders_row_staging;
select * from orders_final_table;
select SYSTEM$STREAM_HAS_DATA('orders_stream');
CREATE OR REPLACE TASK ts_combine_data_changes
WAREHOUSE = COMPUTE_WH
SCHEDULE = '1 MINUTE'
WHEN SYSTEM$STREAM_HAS_DATA('orders_stream')
AS
merge into ORDERS_FINAL_TABLE F -- Target table to merge changes from source table
USING (
select sm.*,b.customer_name,b.email, b.phone
from ORDERS_STREAM sm inner join customers b on sm.customer_id = b.customer_id
) S
ON F.ORDER_ID=S.ORDER_ID
when matched -- DELETE condition
and S.METADATA$ACTION ='DELETE'
and S.METADATA$ISUPDATE = 'FALSE'
then delete
when matched -- UPDATE condition
and S.METADATA$ACTION ='INSERT'
and S.METADATA$ISUPDATE = 'TRUE'
then update
set f.ORDER_DATE = s.ORDER_DATE,
f.STATUS= s.STATUS,
f.TOTAL_AMOUNT=s.TOTAL_AMOUNT
when not matched
and S.METADATA$ACTION ='INSERT'
and S.METADATA$ISUPDATE = 'FALSE'
then insert
(order_id ,order_date ,status ,total_amount, customer_name, email, phone)
values
(s.order_id ,s.order_date ,s.status ,s.total_amount, s.customer_name, s.email, s.phone);
ALTER TASK ts_combine_data_changes RESUME;
SHOW TASKS;
// Change data
INSERT INTO orders_row_staging (order_id, customer_id, order_date, status, total_amount) VALUES
(308, 101, '2024-05-06', 'Pending', 799.98),
(309, 102, '2024-05-07', 'Delivered', 1099.98);
UPDATE orders_row_staging
SET STATUS ='Delivered'
WHERE ORDER_ID = '301';
DELETE FROM orders_row_staging
WHERE ORDER_ID = '302';
// Verify results
select * from orders_stream;
select * from orders_row_staging;
select * from orders_final_table;
// Verify the history
select *
from table(information_schema.task_history())
order by name asc,scheduled_time desc;