External Stages

When you create an external stage in Snowflake, you can think of it like a pointer to a third-party cloud storage location. This can be Amazon S3, Google Cloud Storage, or Microsoft Azure regardless of what platform you run your Snowflake account on. So, for example, you can run your Snowflake account on Amazon and create an external stage on Google Cloud Storage or Microsoft Azure.

When you create an external stage in Snowflake, it creates an object within the selected schema. This object holds the URL along with any required settings to access the cloud storage location. Before creating a stage, you must also define a file format, which tells Snowflake what kind of files to expect. You will use an external stage in the practical example at the end of this chapter.

External Tables and Data Lakes

Within your external stage you can have external tables. These objects hold metadata that tells Snowflake where to locate the data files that relate to the table. This approach allows data to sit outside of Snowflake but appear to users as if it resides within Snowflake.

This can be advantageous if you have a large amount of data in cloud storage but the value of a large proportion of it is yet to be determined. Figure 2-1 illustrates a common pattern I see. A data lake is used to store high volumes of raw data cost effectively, while only a subset of high quality, refined data is loaded into Snowflake. This can be useful when you want your data lake to remain as the source of truth. It allows you to expose a subset of key business data and model it into a format to support the business needs. This approach aims to ensure that the effort required to create to guarantee the delivery of the data to the business is aligned with the data that provides high value.

Meanwhile, the majority of the data residing in the data lake can support exploratory analytics. This looks to take a hypothesis from the business and prove it using the data in the data lake.

For example, let’s say a home insurance product is losing money because the staff didn’t accurately price premiums for homes in an area subject to regular flooding. An operating theory is that if they overlay those areas at risk of flooding based on historical information against their existing pricing structures, they should be able mitigate some of their risk by increasing premiums accordingly.

They want to run some analysis to prove (or disprove) their theory. This is where the data lake can help support the required testing. If the theory ends up being correct, at this point the data can be refined and ingested into the data warehouse. This allows the business to work with the new data points to adapt the existing pricing models to refine the premiums charged to certain customers.

Some organizations I have worked for in the past have insisted on having a centralized data lake to store all raw data. The rationale behind this was to provide maximum flexibility should requirements arise that might not be a natural fit for Snowflake. These requirements could be satisfied using another tool along with data in the data lake, for example.

Other clients view a data lake as a layer of protection against “vendor lock-in” so they avoid placing all their data directly into Snowflake. However, you can actually unload data from Snowflake, which I’ll cover later in this chapter.

The key takeaway here is that you don’t need to load data into Snowflake before you can query it. If your data resides in cloud storage, you can work with it in external tables.

Internal Stages

Whereas external stages focus on data outside of Snowflake, internal stages focus on data within Snowflake. There are three types of internal stages: user, table, and named.

COPY INTO Syntax

Transformations

File Preparation

You’ll want to take advantage of Snowflake’s ability to load data in parallel. So, it is important to consider the size of files to be loaded. They shouldn’t be too large or too small. Snowflake recommends you should aim to produce files of around 100-250MB compressed. This may force you to split larger files into smaller ones or group together several small files into a larger one.

I worked on a project where we needed to migrate a lot of historical data from a legacy on-premise warehouse to Snowflake. To do this, we used our Extract Transform Load (ETL) tool to extract data in monthly increments before we wrote the data out into multiple CSV files. We pushed these files into our cloud storage area (in this case we were using the external stage approach described above). We were then able to load these files in parallel into one target table.

When working with CSV files (or any delimited text files, for that matter), I have found that it is useful to standardize some of this data before loading into Snowflake. Where possible, encode the file in the UTF-8 format, which is the default character set in Snowflake. We also chose to escape any single or double quotes in the data; we also selected unique delimiter characters to be extra sure.

Carriage returns appeared in some of our source data, which were introduced from free text fields in the front-end application it supported. For our specific purposes, we were able to remove them as part of the process.

Semistructured Data

All semistructured data in Snowflake is held in a VARIANT column in the target table. I discussed in Chapter 1 how Snowflake will attempt to treat semistructured data in the same way as relational data behind the scenes by compressing and storing data into a columnar format.

Not every element is extracted into this format. There are some exceptions. One of the most notable is the handling of NULL values. When a “null” value in semistructured data is encountered, Snowflake doesn’t extract that null into a column. The reason for this is so the optimizer can distinguish between a SQL NULL and an element with the value “null.”

Due to the difference in data types used for the same field, Snowflake cannot store these values in an optimized way, resulting in a price to pay. That price is performance. If Snowflake cannot find the extracted element it needs within the columnar format, then it must go back and scan the entire JSON structure to find the values.

To provide Snowflake with the best chance of optimizing the storage of semistructured data, there are a couple of things you can do. If you need to preserve the “null” values, you can extract them into a relational format before you load the data into Snowflake. This creates additional upfront work to initially create the relational structure but the performance gains when querying this data in Snowflake may well be worth it. You’ll have to make this assessment when considering how often this data will be queried.

Alternatively, if the “null” values don’t provide any additional information or context, then you can set the STRIP_NULL_VALUES setting to TRUE in the file format.

Dedicated Virtual Warehouses

When loading data into Snowflake a key concept to understand is how you can use dedicated VWs to prevent contention. You might have had experiences where a user executes an intensive query before going home for the evening. This query continues to run throughout the night and locks tables, thus preventing the ETL process from running. There were measures you could have taken to prevent this from occurring, of course, but I am using this example to illustrate that the design in Snowflake is different and by nature you can easily avoid running into similar issues.

It is common to have dedicated VWs to supply the resources for your data pipelines and bulk data loads. This ensures that the correct amount of resources is available at the right time to allow for efficient loading of data into Snowflake. Configuring multiple dedicated warehouses to line up with business groups, departments, or applications protects these processes from fighting for the same resources. It also allows you to configure each dedicated warehouse in a specific way to match the demand. Figure 2-2 shows an example setup.

Partitioning Staged Data

One thing that is often overlooked at the outset of a project when planning to load data is the partitioning of data in the external cloud storage locations. Essentially, this means separating the data into a logical structure within AWS S3, Google Cloud Storage, or Azure Blob containers using file paths.

Partitioning not only makes it easier to locate files in the future but is important for performance purposes. Creating a structure that includes the source system or application name along with the date the data was written should be considered a minimum requirement. This allows you to pinpoint the data to be loaded onto Snowflake using the COPY command. Figure 2-3 shows an example of physical tables mapped to logical partitions within an S3 bucket.

You will almost certainly want to consider more granular levels of partitioning when it comes to the date. For example, a common approach is to use year/month/day, which can be broken down even further to hour or minute increments. The choice you make will be based on how many files you plan to generate over a certain time period. The goal is to reduce the number of files sitting in each directory as the COPY statement will read a directory list provided by the cloud storage area.

Loading Data

As you load data into Snowflake using the COPY INTO command , metadata is tracked and stored for 64 days. This includes several fields for each data file loaded, such as the name of the file, the file size, and the number of the rows in the file. For a complete list, refer to the official Snowflake documentation at https://docs.snowflake.com/en/user-guide/data-load-considerations-load.html#load-metadata.

Not only is this metadata very useful for understanding what files were loaded when, but it also prevents the same data files being loaded more than once.

In certain environments, you may have limited control over the data files being delivered to you. In these cases, you may not be able to guarantee that the data files aren’t archived after you’ve loaded them. Also, there could be occasions when the same files are mistakenly delivered to the staging area again. In these instances, you don’t need to worry about writing defensive code to check for this. If you’ve loaded the file in the previous 64 days, Snowflake will simply ignore it. Snowflake calculates and stores an MD5 checksum across all records from when the files are loaded. The checksum is stored in the load metadata and subsequently leveraged to prevent the reloading of duplicate files within the 64-day window.

You can override this behavior, of course, and this can be useful for testing. Imagine you want to reload the same file again and again to test an end-to-end process. In this case, you can use the FORCE copy option to override the default behavior.

Loading Using the Web UI

It is also possible to load data via the web UI in Snowflake. Due to several limitations, it’s not something you would use as an enterprise scale process, but rather for simple ad-hoc requests. More typically, non-technical users such as data analysts who are working with small volumes of data may prefer this user-friendly way of loading data into Snowflake so they can carry out analysis on the data. It should go without saying that for regular, operational data loads, this should be incorporated into your ETL process, which will use the COPY INTO command.

Change Tracking Using Streams

Suppose you are loading data into a staging table using Snowpipe. Let’s say you want to track changes on that table in order to merge those incremental changes into a target table.

A stream object tracks any DML operations (inserts, updates, and deletes) against the source table. This process, known as Change Data Capture(CDC), isn’t new but has become far easier to implement over time. One of the primary benefits of using CDC is to help streamline the movement of data. For example, if you have a very large transaction table in your source system containing millions and millions of records, with 1,000 new transactions being added per day, yet no records change historically, you wouldn’t want to reload the entire table each day. Using CDC against this table allows you to identify and target just those 1,000 records for extraction. This makes the entire process faster and more efficient.

In Snowflake, when you create a stream, a few things happen. Firstly, a pair of hidden columns are added to the stream and they begin to store change tracking metadata. At the same time, a snapshot of the source table is logically created. This snapshot acts as a baseline, so that all subsequent changes on the data can be identified. This baseline is referred to as an offset . You can think of it like a bookmark, which stores a position against a sequence of changes. The stream also creates a change table, which stores both the before and after record between two points in time. This change table mirrors the structure of the source table along with the addition of some very handy change tracking columns. You can then point your processes to this change table and process the changed data.

Multiple queries can query the same changed records from a stream without changing the offset. It is important to note that the offset is only moved when stream records are used within a DML transaction. Once the transaction commits, the offset moves forward, so you cannot reread the same record again.

Stream Metadata Columns

Tasks

In this chapter, you’ve learned that using the COPY INTO command allows for batch loads of data, while Snowpipe makes it easy to ingest data into Snowflake on a near real-time basis into a staging table. Adding streams allows for identifying changed records.

Theoretically then, this makes the data available to be consumed directly from the staging table. However, it is more than likely you’ll have to join the incoming stream of data to data from other sources and carry out transformations to make sense of the data before making it available to your end users. You’ll probably have a sequence of steps you need to apply to a specific order. You’ll also want to automate this and make it easy to schedule the execution of these steps. This is where tasks enter the frame.

Tasks execute a single SQL statement, giving you the ability to chain together a series of tasks and dependencies so you can execute them as required.

A typical pattern is to have a task running every 10 minutes, which checks the stream for the presence of records to be processed using the system function SYSTEM$STREAM_HAS_DATA('<stream_name>'). If the stream function returns FALSE, there are no records to process and the task will exit. If the stream function returns TRUE, this means that there are new records to be consumed. In this case, the task contains SQL logic that applies transformations or uses stored procedures or user-defined functions before merging those changes into a target table. This data is then ready to be used in dashboards, analytical models, or be made available to operation systems.

As you’re running DML statements against the data, you’ll need to call on the resource from a virtual warehouse. When you create a task then, you must specify a warehouse to use. Tasks require you to have only one parent task, known as the root task. The root task must also have a schedule associated with it. This can be a duration such as 5 minutes or a CRON expression. It’s worth bearing in mind that as you will automatically be charged for the first 60 seconds of a warehouse starting, setting a duration as low as 5 minutes will incur a significant number of credits. Figure 2-5 provides a flow of a simple chain of tasks to populate a data warehouse.

Full details on all parameters can be found in the official Snowflake documentation.

The Example Scenario

Figure 2-6 illustrates what you’ll put together. In this example, your objective is simply to keep a target table in sync with your source system. This example uses an external stage, the COPY INTO command to unload and load data between the external stage and Snowflake, plus streams and tasks.

Steps

In the following steps you’ll create an Amazon S3 bucket using the AWS console and configure security to allow Snowflake to access the data within the bucket (using an external stage). You’ll then unload some data from a Snowflake sample database to your new S3 bucket. The unloaded data will then be used to simulate data loads into Snowflake, allowing you to use streams and tasks to orchestrate efficient data movement between Snowflake tables.

1. 1.

   Create a new Amazon S3 bucket to act as your external stage, which will store your data. Log into the AWS Console, go to the AWS S3 dashboard, and select the Create Bucket option shown in Figure 2-7.

    

1. 2.

   Provide a name for your bucket and select a region (Figure 2-8).

    

1. 3.

   AWS handily blocks all public access to your newly created S3 bucket. Uncheck the Block all public access check box and tick the box at the bottom of the page to confirm you acknowledge the implications of what you’re about to do (Figure 2-9). Note that you’d NEVER do this in a commercial environment, but as this is just a test using sample data, you’re going to allow all public access to your bucket. I cover how to best manage security and access control in Chapter 4.

    

1. 4.

   Create a folder in your S3 bucket to store the data. Give your folder a name and click Create Folder (Figure 2-10).

    

1. 5.

   Your S3 dashboard view should now look similar to Figure 2-11.

    

1. 6.

   You now need to generate the access keys that Snowflake will require to access the S3 bucket. In the AWS console, go to the Identity and Access Management (IAM) section (Figure 2-12).

    

1. 7.

   Select Access Keys from the main window and click Create New Access Key (Figure 2-13).

    

1. 8.

   Click Show Access Key and make a note of the Key ID and Secret Key. It is important you keep these safe and don’t share them with anyone (Figure 2-14).

    

--CREATE DATABASE

CREATE OR REPLACE DATABASE BUILDING_SS;

--SWITCH CONTEXT

USE DATABASE BUILDING_SS;

--CREATE SCHEMAS

CREATE SCHEMA STG;

CREATE SCHEMA CDC;

CREATE SCHEMA TGT;

--CREATE SEQUENCE

CREATE OR REPLACE SEQUENCE SEQ_01

START = 1

INCREMENT = 1;

--CREATE STAGING TABLE

CREATE OR REPLACE TABLE STG.CUSTOMER

(C_CUSTKEY NUMBER(38,0),

 C_NAME VARCHAR(25),

 C_PHONE VARCHAR(15));

CREATE OR REPLACE TABLE TGT.CUSTOMER

(C_CUSTSK int default SEQ_01.nextval,

  C_CUSTKEY NUMBER(38,0),

 C_NAME VARCHAR(25),

 C_PHONE VARCHAR(15),

 DATE_UPDATED TIMESTAMP DEFAULT CURRENT_TIMESTAMP());

1. 10.

   Create an external stage to point to your S3 bucket. You can choose to do this through the UI by selecting the newly created BUILDING_SS database, selecting Stages, and selecting the Create button. In the next screen, select Existing Amazon S3 Location (Figure 2-15). Then complete the required details.

    

--GRANT PERMISSIONS ON STAGE

GRANT USAGE ON STAGE S3_STAGE TO SYSADMIN;

--SHOW STAGES

SHOW STAGES;

1. 13.

   You should see results similar to Figure 2-16.

    

--UNLOAD DATA TO S3 EXTERNAL STAGE

COPY INTO @S3_STAGE/Customer FROM "SNOWFLAKE_SAMPLE_DATA"."TPCH_SF1"."CUSTOMER"

HEADER=TRUE;

1. 15.

   Refreshing the S3 management console and you should see something similar to Figure 2-17.

    

--SHOW STREAMS

SHOW STREAMS;

--CHECK CHANGE TABLE FOR METADATA COLUMNS

SELECT *

FROM CDC.CUSTOMER;

SELECT * FROM CDC.CUSTOMER;

CREATE OR REPLACE TASK CDC.MERGE_CUSTOMER

  WAREHOUSE = COMPUTE_WH --YOU MUST SPECIFY A WAREHOUSE TO USE

  SCHEDULE = '5 minute'

WHEN

  SYSTEM$STREAM_HAS_DATA('CDC.CUSTOMER')

AS

MERGE INTO TGT.CUSTOMER TGT

USING CDC.CUSTOMER CDC

ON TGT.C_CUSTKEY = CDC.C_CUSTKEY

WHEN MATCHED AND METADATA$ACTION = 'INSERT' AND METADATA$ISUPDATE = 'TRUE'

THEN UPDATE SET TGT.C_NAME = CDC.C_NAME, TGT.C_PHONE = CDC.C_PHONE

WHEN NOT MATCHED AND METADATA$ACTION = 'INSERT' AND METADATA$ISUPDATE = 'FALSE' THEN

INSERT (C_CUSTKEY, C_NAME, C_PHONE) VALUES (C_CUSTKEY, C_NAME, C_PHONE)

WHEN MATCHED AND METADATA$ACTION = 'DELETE' AND METADATA$ISUPDATE = 'FALSE' THEN

DELETE;

--BY DEFAULT A TASK IS SET UP IN SUSPEND MODE

SHOW TASKS;

--ENSURE SYSADMIN CAN EXECUTE TASKS

USE ROLE accountadmin;

GRANT EXECUTE TASK ON ACCOUNT TO ROLE SYSADMIN;

--YOU NEED TO RESUME THE TASK TO ENABLE IT

ALTER TASK CDC.MERGE_CUSTOMER RESUME;