Snowflake

Mirror is Goldsky’s previous-generation pipeline product. If you’re building a new pipeline, use Turbo instead. See the Mirror vs. Turbo comparison, or the migration guide if you have an existing Mirror pipeline.

Mirror doesn’t yet support Snowflake as a native sink. Streaming data directly into Snowflake requires an always-on Snowflake warehouse, which is typically far more expensive than batch loading for the same data.Instead, this page outlines how to load data into Snowflake periodically from S3, a cost-effective storage sink that is natively supported by Goldsky and by Snowflake.

What you’ll need

This page outlines the core process with illustrative scripts (using python and SQL); you’ll need to schedule these tasks into an orchestration platform such as Dagster or Airflow to run on your desired schedule. You’ll also need:

AWS Key ID and AWS Secret Key to access your S3 bucket.
Snowflake account details (account, username, password, database name, warehouse name)
A Goldsky Mirror pipeline writing the desired data to an S3 bucket.

Write extraction script

Install required Python libraries

You will need the pandas library to handle data and the snowflake-connector-python library to connect to Snowflake. Install them using pip:

pip install pandas snowflake-connector-python

Define AWS credentials

Define your AWS Key ID and AWS Secret Key as variables. These will be used to access your S3 bucket.

AWS_KEY_ID = "INSERT_HERE"
AWS_SECRET_KEY = "INSERT_HERE"

Write SQL scripts

Write functions that generate SQL scripts to create a table in Snowflake (if it doesn’t exist) and load data from the S3 bucket into the table.

def get_data_script(chain):
return f"""
create table if not exists YOUR_DATABASE.YOUR_SCHEMA.raw_{chain}data_parquet(DATA_RAW VARIANT);
COPY INTO YOUR_DATABASE.YOUR_SCHEMA.raw{chain}data_parquet
FROM s3://your_bucket/{chain}/data credentials=(AWS_KEY_ID='{AWS_KEY_ID}' AWS_SECRET_KEY='{AWS_SECRET_KEY}')
PATTERN = '..parquet'
FILE_FORMAT = (
TYPE = 'parquet'
);
"""

Connect to Snowflake and execute SQL scripts

Connect to Snowflake using the snowflake.connector.connect function. Then, execute the SQL scripts using the cursor.execute method, and then close the connection.

import snowflake.connector

# Connect to Snowflake
con = snowflake.connector.connect(
user='YOUR_USERNAME',
password='YOUR_PASSWORD',
account='YOUR_ACCOUNT',
warehouse='YOUR_WAREHOUSE',
database='YOUR_DATABASE',
schema='YOUR_SCHEMA'
)

# Create a cursor object
cur = con.cursor()

# Execute the SQL script
cur.execute(get_data_script("your_chain"))

# Close the connection
con.close()

Load data from staging table to destination table

You now have the raw data from S3 loaded into Snowflake; but because S3 as a sink is not re-org aware, you’ll need an intermediate step to manage duplicate/stale data. This can be done with a QUALIFY statement, depending on your specific situation.For example:

{{ config(materialized="table") }}
with
    MATERIALIZED_TABLE as (
        select
            *
        from {{ source("DATABASE", "RAW_S3_TABLE") }}
    )
select *
from MATERIALIZED_TABLE
qualify row_number() over (partition by id order by block_timestamp desc) = 1

The final line in the code above contains the key de-duplication logic based on id.

Can’t find what you’re looking for? Reach out to us at support@goldsky.com for help.

Mirror

Guides

Reference

What you’ll need

Write extraction script

​What you’ll need

​Write extraction script

What you’ll need

Write extraction script