DEV Community: Raj Das

Data Ingestion in Snowflake with Google Cloud Storage - Part I

Raj Das — Fri, 03 May 2024 18:26:11 +0000

In this article we will go through the process to ingest data into Snowflake from Google Cloud Storage (GCS). Along the way we will understand the required concepts and tasks involved in a step by step manner.

You can refer to the official docs here.

Assumptions

You have an active Google Cloud account with permissions to create IAM roles, GCS buckets, Pub/Sub topics.
You have an active Snowflake account with permissions to create Database, Schema, Stage, Integration Objects and fire queries.

Data Source: Open Data

Create a bucket in GCS (GCP)
We create a bucket in GCS snowflake_gcs-stage-bucket and a folder raw_csv inside it to stage the files that we would load into Snowflake later.

For the steps described here, ACCOUNTADMIN user and COMPUTE_WH Warehouse were used in the Snowflake account.

Create a Database (Snowflake)
Here we are creating a database to hold the manufacturing details of various Breweries.



CREATE DATABASE manuf_db;

USE manuf_db;

Create Storage Integration Object (Snowflake)
A storage integration is a Snowflake object. It stores a generated identity and access management (IAM) entity for the external cloud storage location, in our case it will be Google Cloud Storage (GCS). The IAM entity will be provided with permissions to access the GCS location within the GCP Project.

In the below SQL, we are creating the storage integration object which will have access to two locations viz snowflake_gcs-stage-bucket/ and snowflake_gcs-stage-bucket/raw_csv/. This object will enable us to create a staging location to host our files which are located externally i.e. outside of Snowflake.



CREATE STORAGE INTEGRATION snowflake_gcp_int
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = 'GCS'
  ENABLED = TRUE
  STORAGE_ALLOWED_LOCATIONS = ('gcs://snowflake_gcs-stage-bucket/', 'gcs://snowflake_gcs-stage-bucket/raw_csv/');

In case you get the below error, make sure the gcs:// url prefix is preceding the GCS bucket/folder location.



DESC STORAGE INTEGRATION snowflake_gcp_int;

Make a note of the STORAGE_GCP_SERVICE_ACCOUNT from the output above. This is a service account created by Snowflake which needs to be provided with appropriate permissions in Google Cloud so that it can access GCS.
You may notice that the nomenclature of the Service Account appears to be something within an AWS account rather than GCP. This is because my Snowflake account is hosted in AWS.

The STORAGE_GCP_SERVICE_ACCOUNT needs to be provided with the following permissions.
storage.buckets.get storage.objects.create storage.objects.delete storage.objects.get storage.objects.list

Create an IAM Role (GCP)
Create a Role in Google Cloud and add the required permissions. This step can be performed using GCP Web UI or CLI. The gcloud CLI command shown below creates a new role snowflake_gcp_role and adds the required permissions to it. Replace the GCP-PROJECT-ID with your own GCP Project ID in the below command.



gcloud iam roles create snowflake_gcp_role --project=GCP-PROJECT-ID  
 --description="This role will be assigned to Snowflake Account to enable GCS Access" --permissions=storage.buckets.get,storage.objects.create,storage.objects.delete,storage.objects.get,storage.objects.list

Attach Principal and assign Role (GCP)
In this step we will add the STORAGE_GCP_SERVICE_ACCOUNT to the GCS bucket and assign the Role to the account. In the GCP Web UI, go to Cloud Storage and open the GCS bucket we created for staging our files. Click on the Permissions tab.

Now click on Grant Access

Under Add Principals set the STORAGE_GCP_SERVICE_ACCOUNT name and under Role set newly created IAM Role in GCP.

Create an external stage (Snowflake)
Here we are creating an external stage in Snowflake which references data files stored in a location within GCS. We provide the STORAGE_INTEGRATION object name created earlier which has the required access permissions to the external GCS location.



CREATE SCHEMA stages;

CREATE STAGE manu_db.stages.snowflake_gcp_stage
  URL = ('gcs://snowflake_gcs-stage-bucket/raw_csv/')
  STORAGE_INTEGRATION = snowflake_gcp_int;

SHOW STAGES IN manuf_db.stages;

List files from the external stage (Snowflake)
The LIST command may return empty results if no files are loaded in the designated GSC bucket/folder.



LIST @manuf_db.stages.snowflake_gcp_stage;

The LIST command may throw up an error as below if the STORAGE_GCP_SERVICE_ACCOUNT was not provided with appropriate permissions on the GCS bucket/folder.

Once you load files in the GCS bucket/folder, you will see output similar to the below.

You may query the file in the external storage using the below SQL.



SELECT $1, $2, $3 
FROM @manuf_db.stages.snowflake_gcp_stage/beer-manufacturers-microbrewers-brands.csv;

Notice that the Header row is showing up as part of content from the staged file. We need to skip the header to avoid this error. We can create and re-use a File Format object for this.

Create a file format (Snowflake)
A file format is a named object in Snowflake which holds information about a particular file format like CSV/JSON etc. and certain attributes associated with the format. This object comes handy to describe the files in the staged location while loading into Snowflake tables.



CREATE SCHEMA file_formats;

CREATE OR REPLACE FILE FORMAT manuf_db.file_formats.file_format_csv
TYPE = CSV
FIELD_DELIMITER = ','
SKIP_HEADER = 1
FIELD_OPTIONALLY_ENCLOSED_BY = '"';

SHOW FILE FORMATS IN manuf_db.file_formats;

Create Table (Snowflake)
A table is created in the manuf_db. The publicschema is the default schema in Snowflake.



CREATE TABLE manuf_db.public.manuf_tbl(
    manuf_type  varchar(20),
    manuf_name  varchar(100),
    manuf_brand varchar(100)
);

SELECT * FROM manuf_db.public.manuf_tbl;

Loading data into Table (Snowflake)
The COPY INTO command will load staged file, the data from the file beer-manufacturers-microbrewers-brands.csv, into the existing table manuf_tbl by referencing the stage location snowflake_gcp_stage created earlier. Multiple runs of the COPY INTO table will not duplicate the data unless FORCE = TRUE is specified as an optional clause.



COPY INTO manuf_db.public.manuf_tbl
FROM @manuf_db.stages.snowflake_gcp_stage
FILE_FORMAT = manuf_db.file_formats.file_format_csv
FILES = ('beer-manufacturers-microbrewers-brands.csv');

SELECT COUNT(*) FROM manuf_db.public.manuf_tbl;

These are all the Snowflake objects that we created in our account.

Clean Up (Snowflake)
Clean up by deleting the data from the table and also removing the staged file from GCS.



DELETE FROM manuf_db.public.manuf_tbl;

REMOVE @manuf_db.stages.snowflake_gcp_stage PATTERN='.*.csv';

In Part-II of Data Ingestion, we will use SnowPipe to auto-load data in Snowflake.

Kindly let me know if this article was helpful. Your feedback is highly appreciated.

Data Ingestion in Snowflake with Google Cloud Storage - Part II

Raj Das — Fri, 03 May 2024 18:23:57 +0000

In this article we will go through the process to auto ingest data into Snowflake from Google Cloud Storage (GCS) using a Snow Pipe. Along the way we will understand the required concepts and tasks involved in a step by step manner.

You can refer to the official docs here.

Assumptions

You have an active Google Cloud account with permission to create IAM roles, GCS buckets, Pub/Sub topics.
You have an active Snowflake account with permission to create Database, Schema, Stage and Integration Objects.

Data Source: Open Data

A Snowpipe is an event-based data ingestion tool that speeds up the process of loading data from files as soon as they arrive at a designated staging location. We will use Google Cloud Storage (GCS) in this example to load data into Snowflake. Google PubSub will be used to notify the Snowpipe as soon as a file is placed in GCS.

As a pre-requisite, refer to the Part-I article and create all the objects mentioned there before continuing further.

Create a Topic (GCP)
We need to create a Topic in Google PubSub. A message will be published to this topic whenever a new file is uploaded into the GCS bucket/folder.

Here we are creating a topic snowpipe_gcs_topic. The gcloud CLI command below will create the topic if it does not already exist. It will also verify that the GCS bucket has permission to publish events to this topic and grant the required permission if necessary. The OBJECT_FINALIZE event is sent out when a new object is successfully created in the bucket.



gcloud storage buckets notifications create -f json gs://snowflake_gcs-stage-bucket -t snowpipe_gcs_topic -e OBJECT_FINALIZE

gcloud pubsub topics describe snowpipe_gcs_topic

PubSub Topic created

Create a Subscription (GCP)
Next we need to create a subscription to the topic so that we get notified whenever a new message is published to the topic. Here we are creating a subscription snowpipe_gcs_sub to the topic snowpipe_gcs_topic



gcloud pubsub subscriptions create snowpipe_gcs_sub --topic=snowpipe_gcs_topic

gcloud pubsub subscriptions describe snowpipe_gcs_sub

Make sure that the subscription state is active

Service Account Permission (GCP)
A Service Account is created by Snowflake when a Storage Integration Object is created. This Service Account needs to be provided with PubSub role.

Please refer to Part-I for more details on how to create a Storage Integration Object and provide appropriate permissions.

Select the Subscription name in PubSub UI. Click on View Permissions.

Click on Add Principal.

Input the Service Account name and add the PubSub Subscriber Role to the Service Account.

Navigate to the Dashboard page in the Cloud Console, and select your project from the dropdown list.

Click the ADD PEOPLE TO THIS PROJECT button.

Add the service account name. From the Select a role dropdown, select Monitoring Viewer.

Click the Save button. The service account name is added to the Monitoring Viewer role.

Create a Notification Integration (Snowflake)
The notification integration references the Pub/Sub subscription in Google Cloud. Snowflake associates the notification integration with a GCS service account created for your Snowflake account. This service account is created when we created the Storage Integration Object.

Replace the GCP-PROJECT-ID in the below command with your own project ID.



CREATE NOTIFICATION INTEGRATION snowflake_gcp_pubsub_int
  TYPE = QUEUE
  NOTIFICATION_PROVIDER = GCP_PUBSUB
  ENABLED = true
  GCP_PUBSUB_SUBSCRIPTION_NAME = 'projects/GCP-PROJECT-ID/subscriptions/snowpipe_gcs_sub';

DESC NOTIFICATION INTEGRATION snowflake_gcp_pubsub_int;

Create a Snow Pipe (Snowflake)
Lets create a Snow Pipe which will auto-ingest data into the Snowflake table as it arrives in the GCS location.



CREATE SCHEMA manuf_db.pipes;

CREATE PIPE manuf_db.pipes.snow_gcs_pipe
AUTO_INGEST = true
INTEGRATION = snowflake_gcp_pubsub_int
AS
COPY INTO manuf_db.public.manuf_tbl
FROM @manuf_db.stages.snowflake_gcp_stage
PATTERN = '.*manufacturers.*'
FILE_FORMAT = manuf_db.file_formats.file_format_csv;



DESC PIPE manuf_db.pipes.snow_gcs_pipe;



SHOW PIPES IN manuf_db.pipes;

Check the Status of the Snow Pipe. It should be in RUNNING status for it to work correctly. Make sure the Service Account is provided with correct permissions in GCP (as described above).



SELECT SYSTEM$PIPE_STATUS('manuf_db.pipes.snow_gcs_pipe');

Upload files in GCS (GCP)
Upload a file in the stage bucket/folder. As soon as the file is loaded a message will be published to the Topic snowpipe_gcs_topic.

Click on the PubSub Subscription and click on the Metrics tab to view the messages published.

The file now should be loaded to the manuf_tbl in Snowflake. It takes around a minute or two for the data ingestion to happen.

You may load multiple files in the GCS bucket and those will be loaded using the Snow Pipe.

Verify Ingested Data (Snowflake)
Run the below SQL to verify if the data is loaded in the table.



SELECT * FROM manuf_db.public.manuf_tbl;



SELECT count(*) FROM manuf_db.public.manuf_tbl;

The Copy History of the files can also be seen in Snowflake.

Below is the output from the Pipe Status. This data can be very handy for debugging purposes. It displays the last file that was ingested along with the timestamp.



SELECT SYSTEM$PIPE_STATUS('manuf_db.pipes.snow_gcs_pipe');

All Objects created in Snowflake Account Part-I and Part-II)

Fun Fact
Based on the data used, as of 2020-11-01 there are 543 Beer Manufacturer in ON producing 1403 different kinds of Beers.

Kindly let me know if this article was helpful. Your feedback is highly appreciated.