Scribd
Upload
8 views6 pages

Snowflake Continuous Data Pipeline Guide

The document provides examples of continuous data pipelines, detailing the prerequisites for executing SQL statements and showcasing practical use cases such as transforming loaded JSON data, unloading data on a schedule, and refreshing external table metadata. It includes SQL commands for creating tables, streams, and tasks, along with instructions for managing data flow and ensuring timely updates. The examples highlight the importance of access control privileges and the functionality of Snowflake's data pipeline capabilities.

Uploaded by

jitechau01
Copyright
© All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as DOCX, PDF, TXT or read online on Scribd
8 views6 pages

Snowflake Continuous Data Pipeline Guide

The document provides examples of continuous data pipelines, detailing the prerequisites for executing SQL statements and showcasing practical use cases such as transforming loaded JSON data, unloading data on a schedule, and refreshing external table metadata. It includes SQL commands for creating tables, streams, and tasks, along with instructions for managing data flow and ensuring timely updates. The examples highlight the importance of access control privileges and the functionality of Snowflake's data pipeline capabilities.

Uploaded by

jitechau01
Copyright
© All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as DOCX, PDF, TXT or read online on Scribd

Continuous Data Pipeline Examples

This topic provides practical examples of use cases for data pipelines.
Prerequisites
The role used to execute the SQL statements in these examples requires the
following access control privileges:
EXECUTE TASK
Global EXECUTE TASK privilege to run tasks
USAGE
USAGE privilege on the database and schema in which the SQL statements are
executed, as well as on the warehouse that runs any tasks in these examples.
CREATE object
Various CREATE object privileges on the schema in which the SQL statements are
executed, to create objects such as tables, streams, and tasks.
For more information about access control in Snowflake, see Overview of Access
Control.
Transforming Loaded JSON Data on a Schedule
The following example loads raw JSON data into a single landing table
named raw. Two tasks query table streams created on the raw table and insert
subsets of rows into multiple tables. Because each task consumes the change
data capture records in a table stream, multiple streams are required.
-- Create a landing table to store raw JSON data.
-- Snowpipe could load data into this table.
create or replace table raw (var variant);

-- Create a stream to capture inserts to the landing table.


-- A task will consume a set of columns from this stream.
create or replace stream rawstream1 on table raw;

-- Create a second stream to capture inserts to the landing table.


-- A second task will consume another set of columns from this stream.
create or replace stream rawstream2 on table raw;

-- Create a table that stores the names of office visitors identified in the raw
data.
create or replace table names (id int, first_name string, last_name
string);

-- Create a table that stores the visitation dates of office visitors identified in the
raw data.
create or replace table visits (id int, dt date);

-- Create a task that inserts new name records from the rawstream1 stream into
the names table
-- every minute when the stream contains records.
-- Replace the 'mywh' warehouse with a warehouse that your role has USAGE
privilege on.
create or replace task raw_to_names
warehouse = mywh
schedule = '1 minute'
when
system$stream_has_data('rawstream1')
as
merge into names n
using (select var:id id, var:fname fname, var:lname lname from rawstream1)
r1 on [Link] = to_number([Link])
when matched then update set n.first_name = [Link], n.last_name =
[Link]
when not matched then insert (id, first_name, last_name) values ([Link],
[Link], [Link])
;

-- Create another task that merges visitation records from the rawstream2
stream into the visits table
-- every minute when the stream contains records.
-- Records with new IDs are inserted into the visits table;
-- Records with IDs that exist in the visits table update the DT column in the
table.
-- Replace the 'mywh' warehouse with a warehouse that your role has USAGE
privilege on.
create or replace task raw_to_visits
warehouse = mywh
schedule = '1 minute'
when
system$stream_has_data('rawstream2')
as
merge into visits v
using (select var:id id, var:visit_dt visit_dt from rawstream2) r2 on [Link] =
to_number([Link])
when matched then update set [Link] = r2.visit_dt
when not matched then insert (id, dt) values ([Link], r2.visit_dt)
;

-- Resume both tasks.


alter task raw_to_names resume;
alter task raw_to_visits resume;

-- Insert a set of records into the landing table.


insert into raw
select parse_json(column1)
from values
('{"id": "123","fname": "Jane","lname": "Smith","visit_dt": "2019-09-17"}'),
('{"id": "456","fname": "Peter","lname": "Williams","visit_dt": "2019-09-17"}');

-- Query the change data capture record in the table streams


select * from rawstream1;
select * from rawstream2;

-- Wait for the tasks to run.


-- A tiny buffer is added to the wait time
-- because absolute precision in task scheduling is not guaranteed.
call system$wait(70);
-- Query the table streams again.
-- Records should be consumed and no longer visible in streams.

-- Verify the records were inserted into the target tables.


select * from names;
select * from visits;

-- Insert another set of records into the landing table.


-- The records include both new and existing IDs in the target tables.
insert into raw
select parse_json(column1)
from values
('{"id": "456","fname": "Peter","lname": "Williams","visit_dt": "2019-09-25"}'),
('{"id": "789","fname": "Ana","lname": "Glass","visit_dt": "2019-09-25"}');

-- Wait for the tasks to run.


call system$wait(70);

-- Records should be consumed and no longer visible in streams.


select * from rawstream1;
select * from rawstream2;

-- Verify the records were inserted into the target tables.


select * from names;
select * from visits;
Unloading Data on a Schedule
The following example unloads the change data capture records in a stream into
an internal (i.e. Snowflake) stage.
-- Use the landing table from the previous example.
-- Alternatively, create a landing table.
-- Snowpipe could load data into this table.
create or replace table raw (id int, type string);
-- Create a stream on the table. We will use this stream to feed the unload
command.
create or replace stream rawstream on table raw;

-- Create a task that executes the COPY statement every minute.


-- The COPY statement reads from the stream and loads into the table stage for
the landing table.
-- Replace the 'mywh' warehouse with a warehouse that your role has USAGE
privilege on.
create or replace task unloadtask
warehouse = mywh
schedule = '1 minute'
when
system$stream_has_data('RAWSTREAM')
as
copy into @%raw/rawstream from rawstream overwrite=true;
;

-- Resume the task.


alter task unloadtask resume;

-- Insert raw data into the landing table.


insert into raw values (3,'processed');

-- Query the change data capture record in the table stream


select * from rawstream;

-- Wait for the tasks to run.


-- A tiny buffer is added to the wait time
-- because absolute precision in task scheduling is not guaranteed.
call system$wait(70);
-- Records should be consumed and no longer visible in the stream.
select * from rawstream;

-- Verify the COPY statement unloaded a data file into the table stage.
ls @%raw;
Refreshing External Table Metadata on a Schedule
The following example refreshes the metadata for an external table
named [Link] (using ALTER EXTERNAL TABLE … REFRESH) on
a schedule.
Note
When an external table is created, the AUTO_REFRESH parameter is set
to TRUE by default. We recommend that you accept this default value for
external tables that reference data files in either Amazon S3 or Microsoft Azure
stages. However, the automatic refresh option is not available currently for
external tables that reference Google Cloud Storage stages. For these external
tables, manually refreshing the metadata on a schedule can be useful.
-- Create a task that executes an ALTER EXTERNAL TABLE ... REFRESH statement
every 5 minutes.
-- Replace the 'mywh' warehouse with a warehouse that your role has USAGE
privilege on.
CREATE TASK exttable_refresh_task
WAREHOUSE=mywh
SCHEDULE='5 minutes'
AS
ALTER EXTERNAL TABLE [Link] REFRESH;

Common questions

Powered by AI

The 'system$wait' function provides buffer time to account for potential scheduling delays. This function is critical in ensuring that scheduled tasks have adequate time to complete before further data operations are executed. It adds precision by introducing intentional pauses, enabling reliable task execution even when task timing lacks absolute precision .

The task resumption command 'alter task ... resume;' reactivates any paused tasks, allowing them to continue executing scheduled operations according to their predefined intervals. This command ensures that data pipelines can resume their processes without manually restarting each component, thereby maintaining workflow continuity and processing efficiency .

Tasks that merge new records into existing tables support real-time integration and coherence of data by efficiently appending or updating existing entries. This approach ensures that databases accurately reflect the most recent data state without redundancy, supports analytical processes with current information, and maintains data integrity across updates .

Configuring a refresh task at a specific interval, like every five minutes, ensures that the metadata of external tables is consistently updated, preventing stale or mismatched data references. This manual schedule is especially crucial for tables linked to storage solutions lacking auto-refresh capabilities, ensuring the external table accurately mirrors the underlying data changes .

Parsing JSON data early in the data pipeline is critical for transforming unstructured or semi-structured data into a structured, analyzable format. This step supports subsequent data operations like querying and processing, enabling efficient data manipulation and integration into relational databases, thereby bridging JSON flexibility with structured table reliability .

Access control ensures secure execution by granting specific privileges required for various operations such as EXECUTE TASK, USAGE, and CREATE on objects within the pipeline. This structured permission system restricts task operations to authorized roles only, preventing unauthorized data access or manipulation .

The stream_has_data function is pivotal when creating tasks because it checks if streams contain new records to process, triggering task execution only when data is present. This efficient function minimizes unnecessary computational resource use, optimizes resource allocation, and ensures that pipelines respond to dynamic data states rather than static timing .

Using multiple streams in the data pipeline enables the separation and specific handling of different subsets of change data capture (CDC) records. This design allows separate tasks to consume different columns from the landing table and insert them into various target tables. For instance, one stream focuses on visitor names and another on visitation dates. This separation facilitates efficient data transformation by enabling each task to handle distinct data processing independent of others .

AUTO_REFRESH, when set to TRUE, simplifies metadata management by automatically updating external table metadata without user intervention, provided it’s referencing AWS S3 or Azure stages. This setting reduces manual maintenance overhead and ensures that up-to-date metadata is available, thus improving data consistency and reducing complexity in environments that auto-update metadata .

The COPY statement enables efficient unloading of change data capture records into a Snowflake stage by automating data transfer from table streams to storage stages. This process supports data archiving and backup functions while ensuring no data loss or redundancy since records are automatically consumed in the process .

You might also like