Creating a mock data pipeline with Snowflake and AWS – part 1

Over the past few years, have you ever found a website which would let you log in without requiring your age, phone number, email address and name of your first pet ? Data has become a pillar stone of any online presence aiming to make each experience as personal as possible. With the increasing quantity of data comes the increasing needs for reliable and efficient data storing and processing. And that’s where Snowflake comes into the picture.

Snowflake is a cloud computing-based data warehouse allowing you to ingest, store, process and export incredible amounts of data in a very short time and with usage-based costs. Just like BigQuery, Redshift and other current cloud-computing data tools, the move to the cloud is a no-brainer as it allows more flexibility, is very easy to set up and use. But with such a great offer, obviously comes some cons: cloud based warehouses are very performant with massive amounts of data but still struggling when it comes to REST API and querying specific rows in this big ocean of data. That’s why older style databases like PostgreSQL or noSQL ones like DynamoDB are not to be forgotten as they are a great serving layer between cloud warehouses and in-app analytics.

This is the first part of a 2-part article aiming to talk about data clustering and partitioning and its importance in data querying efficiency. In this first part, we aim to create a data pipeline with AWS and Snowflake. In the second part we will talk more in depth about performance and data structures.

Context and data pipeline description

For this, we are going to replicate the process of delivering in-app analytics as we do at Linktree. Visitors view and click on Linktree profiles and we give our customers the ability to see their analytics.

For this you will need a Snowflake account which you can trial for free as well as an AWS account also available for free. Let’s get started. Below is the diagram of the objects we are going to build.

At Linktree we love to use Infrastructure as Code with AWS CDK, this allows us to easily maintain code, maintain our data warehouses, deploy new pipelines and make processes faster. Basically all the AWS objects are defined through code instead of through the AWS website, however it is totally possible to do it directly in the website if preferred.

For this use case we are going to use a Glue job to automate mock data, a S3 bucket to store our data and a Snowflake pipe to bring the data from S3 to Snowflake; then mostly using Snowflake objects, like streams and tasks, to move the data automatically across tables.

A Github repository has been created for this article to easily spin up the architecture above, see here https://github.com/clevecque/cdk-snowflake-demo for details.

AWS side

The Glue job represents our web application, for us a Linktree profile. In theory you would define your events in your backend, at Linktree we like to send them all to Eventbridge as it can have several sources of entry. However to make it easier, here it is only a Glue job generating random data on a regular basis. Here is an example of the data generated, the `event_type` value can be `click` or `view`.

{
  "country": "UK",
  "device": "tablet",
  "event_type": "click",
  "id": 90774182,
  "timestamp": "2022-03-14T23:50:36.190139",
  "user_id": 289
}

In the stack definition you can see we use a CRON job to schedule this Glue job to run every 10 minutes. The job generates a set of data and exports it into our storage S3 bucket.

const glueTrigger = new CfnTrigger(this, 'glueJobDataGeneratorTrigger', {
     name: 'etl-trigger',
     schedule: 'cron(0/10 * ? * * *)',
     type: 'SCHEDULED',
     actions: [
       {
         jobName: glueJobDataGenerator.name
       }
     ],
     startOnCreation: true
});
glueTrigger.addDependsOn(glueJobDataGenerator);

You can easily pause or resume your Glue job by going into the Glue Studio in your AWS account, selecting your Glue job and then going into the `Schedules` tab. You can also monitor it more easily through there as we haven’t set any monitoring tool.

Once the data lands into the S3 bucket, it triggers a notification sent to Snowflake, thanks to a SNS subscription, that new data has arrived.

const dataBucketSNSTopic = new Topic(this, `dataBucketSNSTopic`, {
     topicName: `cdk-snowflake-demo-bucket-sns-topic`
   })

Snowflake side

On Snowflake’s side, this notification triggers the pipe to run, automatically pulling the data from the bucket. Every time the pipe receives a notification of new data, the following code is run importing data in a staging table called `events`.

COPY INTO stage.events
       FROM
           (SELECT current_timestamp      AS _load_time
               , metadata$filename        AS _file_name
               , metadata$file_row_number AS _file_row_number
               , sha2($1)                 AS _hash_key
               , $1::variant              AS raw_data
           FROM @stage.stage_external_events
           )
FILE_FORMAT = (type = json);

Since we want this table to act as a stage, we store the data without any processing, this ensures that all data from the S3 bucket will make it to the table.

From the moment the Glue job succeeds, data is almost immediately received in Snowflake’s stage table.

And that’s where we start talking about tasks and streams. A stream is an object monitoring changes in a table, so inserts, updates and deletions, what is called Change Data Capture. By defining a stream on our stage table `events`, we can more easily track which data is new and which data has already been processed. This allows not to scan the entire table just to get the new data. I recommend having a look at Snowflake documentation detailing how it works.

Every 15 min, the task `task_events` looks at the new data in the stream and processes it to store it either in the `clicks` table or the `views` table based on the `event_type` value.

Finally, once this has run, the last set of tasks is being triggered (let’s call them `merge tasks` or `rollups`) aggregating the data on an hourly basis for easier serving to BI or customers for instance.

We specify `merge tasks` to differentiate them from purely `insert tasks`. Indeed the Snowflake operations are different in each case from either an `INSERT` statement or a `MERGE` statement.

We could wait until the end of the hour, aggregate the hourly data and insert it into the hourly table. However let’s say we want to serve the users their analytics as fast as possible. We are in a use case of close to real-time data, so it is not possible to just wait until the end of the hour to compute the hourly analytics. As soon as data is received, it is aggregated which means we might need to update a total count for instance hence the merge operation.

And that’s it, now your data is sanitised and ready to be served to the users in a clearer way. You can decide across the pipeline to add conditions in the tasks or different aggregations or computations.

You can also research it to skip this system of tasks and streams which can easily become overwhelming and replace it with tools like Airflow made to organize and schedule tasks and processes.

Next time, we’ll look at how these cloud computing warehouses and in particular Snowflake are storing, partitioning and clustering the data to make it so easy to use.