Staff Data Engineer Tech Test

Requirements:

• Python 3.13
• uv
• Snowflake Account Free Trial Account

Setup

This project uses uv as a package manager.

pip install uv
uv venv
source .venv/bin/activate
uv sync

Snowflake Set Up

Run the following in a Snowflake account to set up the required resources for running this project (normally this would be terraformed)

-- create the database
use role sysadmin;
create database podcasts;

-- set up for ingestion 

use role sysadmin;
create schema podcasts.raw;
create stage podcasts.raw.raw_data;
create warehouse loader warehouse_size = 'xsmall' auto_suspend = 60;

-- create a loader role 
use role securityadmin;
create role svc_role_loader;
grant role svc_role_loader to role sysadmin;

-- give loader role relevant permissions
grant usage on database podcasts to role svc_role_loader;
grant usage on schema podcasts.raw to role svc_role_loader;
grant usage on warehouse loader to role svc_role_loader;
grant read, write on stage podcasts.raw.raw_data to role svc_role_loader;

-- now create raw tables for data load
use role sysadmin;
use schema podcasts.raw;

create or replace table episodes (
    episode_id string,
    podcast_id string,
    title string,
    release_date date,
    duration_seconds int,
    load_timestamp timestamp default current_timestamp()
);

create or replace table events (
    payload variant,
    load_timestamp timestamp default current_timestamp()
);

create or replace table users (
    user_id string,
    signup_date date,
    country string,
    load_timestamp timestamp default current_timestamp()
);

-- grant access to raw tables
grant insert on all tables in schema podcasts.raw to role svc_role_loader;

use role svc_role_loader;

-- create a role for dbt to use
use role securityadmin;
create role svc_role_dbt;
grant role svc_role_dbt to role sysadmin;

-- give dbt the required permissions
grant usage create schema on database podcasts to role svc_role_dbt;
grant usage on warehouse compute_wh to role svc_role_dbt;
grant usage on schema podcasts.raw to role svc_role_dbt;
grant select on all tables in schema podcasts.raw to role svc_role_dbt;

Snowflake Credentials

Create a service user in Snowflake for each of the service roles created (or use your own user, not for prod, but this is just a hacked together project!) Once created, create a PAT for the user, and then set the following environment variables:

export SNOWFLAKE_USER=xxxxx
export SNOWFLAKE_PASSWORD=xxxxx
export SNOWFLAKE_ACCOUNT=xxxxx

These values are used in ingest/main.py

Running Ingest

The ingest is set up in /ingest. To run, ensure the steps for uv above have been followed, and then run the following from the root directory:

python ingest/main.py

This will load the files in materials/ to an internal stage in Snowflake, and then into the raw tables in PODCASTS.RAW

Running dbt

Create a user in Snowflake with the SVC_ROLE_DBT role (as defined in Snowflake Setup section)

Create a PAT for the user and set the following environment variables:

export SNOWFLAKE_ACCOUNT=xxxxx
export DBT_USERNAME=xxxxx
export DBT_PASSWORD=xxxx

The run the following to test the connection:

cd dbt_podcasts
dbt debug

To run the dbt project, simply run the below:

dbt build

This will run both the models and the associated tests.

Orchestration

Due to time contraints, I've not built out the DAG to run the pipeline. However, it would be something like the below:

 Ingestion

As we're receiving event data in this example, in production it's likely we want to stream this and have near real time processing.

If working with AWS, this would require:

• Subscribing to the events from the upstream application e.g with AWS EventBridge
• Routing events to Kinesis Firehose and batching into AWS S3
• Setting up a Storage Integration object in Snowflake, and associated IAM role and S3 Bucket Policy in AWS.
• Creating a Snowpipe object on Snowflake with auto_ingest set to true.
• Setting up an S3 event notification and routing this to the SQS queue of the Snowpipe object

This pattern allows for easy ingestion of new data sources.

Transformation

The dbt project will need to be run on a schedule. The approach I would take to this would be:

• Use dbt Cloud if money is no object! (Or use the public preview of dbt projects natively in Snowflake)

Or...

• Build the dbt project, including all dependencies, and dbt model into a Docker image. (Do this on merge in a GitHub action pipeline)
• Push the built image to a container registry e.g. AWS ECR.
• Set up an Airflow DAG to run the dbt project on a schedule.
• The DAG should have a task to run the docker image which runs dbt.
• One approach is to use AWS ECS. If doing this, set up a Secrets Manager backend for Airflow to allow it to securely fetch the Snowflake credentials.
• Configure alerting using the Airflow on_failure_callback config. Can send messages to SNS, and then route these SNS messages to Slack using AWS Chatbot.

Part 3 - Best Practices

Modelling Approach

• Creating a staging layer that:
  • Creates a surrogate primary key
  • Uses the key for deduplication. Protects against same file being loaded twice with different names etc.
  • Allows for easy renaming of columns if required.
  • Extracts key fields from variant (JSON) data.
  • Adding a data quality flag, identifying any records that don't conform to the event schema.
• Mart Layer
  • At this point, the model could have been split into facts and dimensions, which would traditionally handle the requirement to store the data efficiently. However, there is minimal dimensional attributes, and so in this case, modelling a mart table with the dimensional data already joined to the event data makes it more efficient for user querying (reducing warehouse spend) with a minimal impact on storage volumes (due to the compressed columnar storage in Snowflake, mitigating impact of storing repeated low cardinality attributes.)
  • Mart layer is built incrementally, reducing transformation load.
  • Filtering out any records with an unacceptable data quality.

Testing Strategy

Python

• Due to time constraints, I've not added unit tests of the ingest code, but I would would mock the SnowflakeConnector class and it's methods, and then assert that the functions are called with the correct arguments.
• This would include asserting the calls passed to the cursor.execute commands, especially in the cases where the SQL is being dynamically generated. Also testing for error handling, e.g. passing of unsupported formats.

dbt

• Source freshness checks added. Pipeline should error out and alert engineers if we've not received any new data without expected timeframe.
• Basic tests added at staging layer, including checking PK assumptions are valid, and relationships are valid.
• Mart layer contains more thorough testing, where it is more performant to execute due to materialised nature.
  • Use of dbt built in tests for relationships, unique, not null, accepted values (also example of not null test with a where condition added)
  • Use of custom generic test for looking at valid timestamps (in this simple example checking all timestamps are within 1000 days). This is easily parameterised.
• Tests are driven with the guidance of the business users, in this case, the provided document.

Handling Malformed Data

• This ingest uses an strict ELT pattern. This means all data is loaded to the warehouse, regardless of what it is. If the event JSON completely changed, it would still be loaded. This ensures we capture everything, and keeps a large amount of flexibility in the process.
• The same is true of the CSV data, where a match by column name method is used.
• This buys us the ultimately flexibility, especially in the case of the JSON data. Malformed data is handled gracefully in this pipeline. However it's not without it's pitfalls. There's a real chance the ingested data will be filled with null values. Hence the importance of the dbt testing process above, with suitable alerting.

Performance

• How the event data is stored will have the most significant impact on performance of this pipeline. It should be partitioned based on when the event occurred, as the query pattern that will hit this data the most will be that of the incremental dbt model, which is filtering based on latest timestamps. Fortunately this is the pattern Snowflake handles by default with micro-partitioning. This can be verified by viewing the number of partitions scanned in the query profile (this small example has one partition, but in a real event dataset it is likely tens of thousands).
• Keeping any BI processes, end users, etc away from the raw events table where possible is paramount. Even simple predicates can result in very long running queries if having to inspect the contents of the variant column.

Reusability

• The ingest pipeline is built around one simple Snowflake connector class that has some useful methods e.g. loading data from a stage to a table.
• It is easily extensible to provide other functionality e.g. deleting files from a stage
• Additional file formats could be added with ease.
• There's a clear pattern for the staging models in dbt, which is easily repeatable for any future data sources.
• Making use of opensource dbt packages e.g. dbt_utils, to reduce developer overhead.
• Making use of Python package manager with locked dependencies (in this case, uv), ensuring reproducible local environments.

Docs

• The developer docs should be added as close to the code as possible e.g. this README file, docstrings on Python functions, comments on dbt models, simple diagrams in Excalidraw, ERD tooling etc.
• Try to reduce the need for engineers to context switch to other tools e.g. Confluence as much as possible, when they are doing simple jobs e.g. fixing a bug in a class method.
• Data docs should be easily accessible to the consumers of this data set. This could be using the likes of dbt docs (hosted by dbt Cloud, or as a static site), or directly in Snowflake.
• This project has persist_docs set to true, so all dbt column and model descriptions are materialised as comments on the tables, views and columns in Snowflake. This is useful as it's easily accessible when people are querying the data.

Governance

• This project contains a simplified but robust approach to RBAC. Separate roles have been created for the loading the data, and for transformation in dbt, each with a very specific set of permissions needed to do their job.
• All the data is loaded, even records with data quality issues. Whilst these records are filtered out in the mart, it is easy to find them at the staging layer and understand what data we are not processing, which can be very useful for audits.