Python Data Loading from shopify
to snowflake
using dlt
Library
Join our Slack community or book a call with our support engineer Violetta.
This page provides technical documentation for using the open-source Python library, dlt
, to load data from Shopify
to Snowflake
. Shopify
is a comprehensive commerce platform that empowers anyone to initiate, expand, manage, and scale a business. On the other hand, Snowflake
is a cloud-based data warehousing platform designed for storing, processing, and analyzing large data volumes. By utilizing dlt
, you can effectively bridge these two platforms, transferring data from your Shopify
store to the Snowflake
warehouse for in-depth analysis and insights. For more information about Shopify
, please visit https://www.shopify.com/.
dlt
Key Features
- Automated maintenance:
dlt
provides automated maintenance with schema inference, evolution and alerts. Your code remains short and declarative, making maintenance tasks simpler. Learn more here. - Scalability and Finetuning:
dlt
offers mechanisms and configuration options to scale up and fine-tune pipelines. It supports running extraction, normalization and load in parallel, and offers options to adjust memory buffers, intermediary file sizes and compression options. Read more about its performance. - Governance Support:
dlt
pipelines provide robust governance support through pipeline metadata utilization, schema enforcement and curation, and schema change alerts. This contributes to better data management practices, compliance adherence, and overall data governance. Learn more here. - Flexible Authentication:
dlt
supports multiple authentication types for Snowflake destination including password authentication, key pair authentication, and external authentication. This flexibility allows for secure and convenient data transfer. Find out more here. - Community Support: The
dlt
community is active and supportive. You can join their Slack to find recent releases or discuss what you can build withdlt
. You can also report problems and make feature requests here.
Getting started with your pipeline locally
0. Prerequisites
dlt
requires Python 3.8 or higher. Additionally, you need to have the pip
package manager installed, and we recommend using a virtual environment to manage your dependencies. You can learn more about preparing your computer for dlt in our installation reference.
1. Install dlt
First you need to install the dlt
library with the correct extras for Snowflake
:
pip install "dlt[snowflake]"
The dlt
cli has a useful command to get you started with any combination of source and destination. For this example, we want to load data from Shopify
to Snowflake
. You can run the following commands to create a starting point for loading data from Shopify
to Snowflake
:
# create a new directory
mkdir shopify_dlt_pipeline
cd shopify_dlt_pipeline
# initialize a new pipeline with your source and destination
dlt init shopify_dlt snowflake
# install the required dependencies
pip install -r requirements.txt
The last command will install the required dependencies for your pipeline. The dependencies are listed in the requirements.txt
:
dlt[snowflake]>=0.3.8
You now have the following folder structure in your project:
shopify_dlt_pipeline/
├── .dlt/
│ ├── config.toml # configs for your pipeline
│ └── secrets.toml # secrets for your pipeline
├── shopify_dlt/ # folder with source specific files
│ └── ...
├── shopify_dlt_pipeline.py # your main pipeline script
├── requirements.txt # dependencies for your pipeline
└── .gitignore # ignore files for git (not required)
2. Configuring your source and destination credentials
The dlt
cli will have created a .dlt
directory in your project folder. This directory contains a config.toml
file and a secrets.toml
file that you can use to configure your pipeline. The automatically created version of these files look like this:
generated config.toml
# put your configuration values here
[runtime]
log_level="WARNING" # the system log level of dlt
# use the dlthub_telemetry setting to enable/disable anonymous usage data reporting, see https://dlthub.com/docs/telemetry
dlthub_telemetry = true
[sources.shopify_dlt]
shop_url = "shop_url" # please set me up!
organization_id = "organization_id" # please set me up!
generated secrets.toml
# put your secret values and credentials here. do not share this file and do not push it to github
[sources.shopify_dlt]
private_app_password = "private_app_password" # please set me up!
access_token = "access_token" # please set me up!
[destination.snowflake.credentials]
database = "database" # please set me up!
password = "password" # please set me up!
username = "username" # please set me up!
host = "host" # please set me up!
warehouse = "warehouse" # please set me up!
role = "role" # please set me up!
2.1. Adjust the generated code to your usecase
3. Running your pipeline for the first time
The dlt
cli has also created a main pipeline script for you at shopify_dlt_pipeline.py
, as well as a folder shopify_dlt
that contains additional python files for your source. These files are your local copies which you can modify to fit your needs. In some cases you may find that you only need to do small changes to your pipelines or add some configurations, in other cases these files can serve as a working starting point for your code, but will need to be adjusted to do what you need them to do.
The main pipeline script will look something like this:
"""Pipeline to load shopify data into BigQuery.
"""
import dlt
from dlt.common import pendulum
from typing import List, Tuple
from shopify_dlt import shopify_source, TAnyDateTime, shopify_partner_query
def load_all_resources(resources: List[str], start_date: TAnyDateTime) -> None:
"""Execute a pipeline that will load the given Shopify resources incrementally beginning at the given start date.
Subsequent runs will load only items updated since the previous run.
"""
pipeline = dlt.pipeline(
pipeline_name="shopify", destination='snowflake', dataset_name="shopify_data"
)
load_info = pipeline.run(
shopify_source(start_date=start_date).with_resources(*resources),
)
print(load_info)
def incremental_load_with_backloading() -> None:
"""Load past orders from Shopify in chunks of 1 week each using the start_date and end_date parameters.
This can useful to reduce the potiential failure window when loading large amounts of historic data.
Chunks and incremental load can also be run in parallel to speed up the initial load.
"""
pipeline = dlt.pipeline(
pipeline_name="shopify", destination='snowflake', dataset_name="shopify_data"
)
# Load all orders from 2023-01-01 to now
min_start_date = current_start_date = pendulum.datetime(2023, 1, 1)
max_end_date = pendulum.now()
# Create a list of time ranges of 1 week each, we'll use this to load the data in chunks
ranges: List[Tuple[pendulum.DateTime, pendulum.DateTime]] = []
while current_start_date < max_end_date:
end_date = min(current_start_date.add(weeks=1), max_end_date)
ranges.append((current_start_date, end_date))
current_start_date = end_date
# Run the pipeline for each time range created above
for start_date, end_date in ranges:
print(f"Load orders between {start_date} and {end_date}")
# Create the source with start and end date set according to the current time range to filter
# created_at_min lets us set a cutoff to exclude orders created before the initial date of (2023-01-01)
# even if they were updated after that date
data = shopify_source(
start_date=start_date, end_date=end_date, created_at_min=min_start_date
).with_resources("orders")
load_info = pipeline.run(data)
print(load_info)
# Continue loading new data incrementally starting at the end of the last range
# created_at_min still filters out items created before 2023-01-01
load_info = pipeline.run(
shopify_source(
start_date=max_end_date, created_at_min=min_start_date
).with_resources("orders")
)
print(load_info)
def load_partner_api_transactions() -> None:
"""Load transactions from the Shopify Partner API.
The partner API uses GraphQL and this example loads all transactions from the beginning paginated.
The `shopify_partner_query` resource can be used to run custom GraphQL queries to load paginated data.
"""
pipeline = dlt.pipeline(
pipeline_name="shopify_partner",
destination='snowflake',
dataset_name="shopify_partner_data",
)
# Construct query to load transactions 100 per page, the `$after` variable is used to paginate
query = """query Transactions($after: String, first: 100) {
transactions(after: $after) {
edges {
cursor
node {
id
}
}
}
}
"""
# Configure the resource with the query and json paths to extract the data and pagination cursor
resource = shopify_partner_query(
query,
# JSON path pointing to the data item in the results
data_items_path="data.transactions.edges[*].node",
# JSON path pointing to the highest page cursor in the results
pagination_cursor_path="data.transactions.edges[-1].cursor",
# The variable name used for pagination
pagination_variable_name="after",
)
load_info = pipeline.run(resource)
print(load_info)
if __name__ == "__main__":
# Add your desired resources to the list...
resources = ["products", "orders", "customers"]
load_all_resources(resources, start_date="2000-01-01")
# incremental_load_with_backloading()
# load_partner_api_transactions()
Provided you have set up your credentials, you can run your pipeline like a regular python script with the following command:
python shopify_dlt_pipeline.py
4. Inspecting your load result
You can now inspect the state of your pipeline with the dlt
cli:
dlt pipeline shopify info
You can also use streamlit to inspect the contents of your Snowflake
destination for this:
# install streamlit
pip install streamlit
# run the streamlit app for your pipeline with the dlt cli:
dlt pipeline shopify show
5. Next steps to get your pipeline running in production
One of the beauties of dlt
is, that we are just a plain Python library, so you can run your pipeline in any environment that supports Python >= 3.8. We have a couple of helpers and guides in our docs to get you there:
The Deploy section will show you how to deploy your pipeline to
Deploy with Github Actions:
dlt
allows you to use Github Actions as a CI/CD runner. You can specify when the GitHub Action should run using a cron schedule expression. Follow the guide on how to deploy a pipeline with Github Actions to learn more.Deploy with Airflow:
dlt
provides an easy way to deploy your pipeline with Airflow. It creates an Airflow DAG for your pipeline script that you should customize. Check out the guide on how to deploy a pipeline with Airflow for more information.Deploy with Google Cloud Functions: With
dlt
, you can also deploy your pipeline with Google Cloud Functions. This allows you to execute your pipeline in response to an event, such as a message on a Pub/Sub topic, a change in a Cloud Storage bucket, or an HTTP request. Visit the guide on how to deploy a pipeline with Google Cloud Functions to learn more.Other Deployment Options:
dlt
offers a variety of other deployment options to suit your needs. Find out more about these options in the deployment guide.
The running in production section will teach you about:
- Monitor Your Pipeline:
dlt
provides tools to help you monitor your pipeline, ensuring that it is running smoothly and efficiently. For more details, check out the guide on how to monitor your pipeline. - Set Up Alerts: With
dlt
, you can set up alerts to notify you of any important events or issues with your pipeline. This allows you to react quickly to any problems and keep your pipeline running smoothly. Learn more about it in the set up alerts guide. - Set Up Tracing:
dlt
also offers tracing capabilities, allowing you to track the execution of your pipeline and identify any potential bottlenecks or issues. Find out more in the set up tracing guide.
Available Sources and Resources
For this verified source the following sources and resources are available
Source shopify
"Shopify is an e-commerce platform offering data on customer accounts, transactions, and product listings."
Resource Name | Write Disposition | Description |
---|---|---|
customers | merge | Individuals or entities who have created accounts on a Shopify-powered online store |
orders | merge | Transactions made by customers on an online store |
products | merge | The individual items or goods that are available for sale |
Additional pipeline guides
- Load data from The Local Filesystem to Databricks in python with dlt
- Load data from PostgreSQL to AWS Athena in python with dlt
- Load data from Chess.com to Azure Synapse in python with dlt
- Load data from Keap to Supabase in python with dlt
- Load data from HubSpot to AWS S3 in python with dlt
- Load data from Chargebee to Supabase in python with dlt
- Load data from IBM Db2 to AWS Athena in python with dlt
- Load data from ClickHouse Cloud to Snowflake in python with dlt
- Load data from Stripe to Timescale in python with dlt
- Load data from Looker to EDB BigAnimal in python with dlt