In this guide, you'll learn how to build a complete machine learning pipeline using Snowflake ML Jobs and Task Graphs. This end-to-end solution demonstrates how to orchestrate the entire ML lifecycle - from data preparation to model deployment - all within Snowflake's ecosystem.

The pipeline you'll build includes data preparation, model training on distributed compute, model evaluation, conditional promotion based on quality metrics, and automated cleanup - all orchestrated through a Task Graph that can be scheduled for recurring execution.

You'll build a complete, production-ready ML pipeline that:

1. Prepares training data from a Feature Store
2. Trains an XGBoost model using distributed compute on Snowpark Container Services
3. Evaluates model quality against predefined thresholds
4. Conditionally promotes high-quality models to production in Model Registry
5. Sends notifications for important events
6. Performs cleanup operations

Firstly, to follow along with this quickstart, you can use the code in the e2e_task_graph GitHub repo to download the code used in this article.

Work with your account administrator to provision the required resources in your Snowflake account as needed.

1. Create a compute pool if you don't already have one:

Note: MAX_NODES should be at least equal to target_instances (2 in this example).

2. Create a virtual warehouse if you don't already have one:

3. Configure database privileges for the demo. Subsequent steps will create resources inside this database.

4. Grant other required access control privileges needed to execute the sample:

5. (Optional) Configure a notification integration to enable sending notifications from Task Graph executions.
   1. Option 1: Email notifications
      1. Verify the email addresses of intended notification recipients
      2. Create the email notification integration

         SET user_email = (SELECT EMAIL FROM SNOWFLAKE.ACCOUNT_USAGE.USERS WHERE NAME = CURRENT_USER());
         CREATE OR REPLACE NOTIFICATION INTEGRATION DEMO_NOTIFICATION_INTEGRATION
             TYPE=EMAIL
             DEFAULT_RECIPIENTS = ($user_email)
             DEFAULT_SUBJECT = 'Model Training Status'
             ENABLED=TRUE;
         GRANT USAGE ON INTEGRATION DEMO_NOTIFICATION_INTEGRATION TO ROLE ENGINEER;
         

   2. Option 2: Webhook notifications (e.g. Slack, Teams)
      1. Create a webhook with your desired notification channel (e.g. Slack Webhook)
      2. Configure notification integration with your webhook

         CREATE SECRET IF NOT EXISTS DEMO_WEBHOOK_SECRET
             TYPE = GENERIC_STRING
             SECRET_STRING = 'T00000000/B00000000/XXXXXXXXXXXXXXXXXXXXXXXX'; -- (ACTION NEEDED) Put your webhook secret here
         CREATE OR REPLACE NOTIFICATION INTEGRATION DEMO_NOTIFICATION_INTEGRATION
             TYPE=WEBHOOK
             ENABLED=TRUE
             WEBHOOK_URL='https://hooks.slack.com/services/SNOWFLAKE_WEBHOOK_SECRET'
             WEBHOOK_SECRET=DEMO_WEBHOOK_SECRET
             WEBHOOK_BODY_TEMPLATE='{"text": "SNOWFLAKE_WEBHOOK_MESSAGE"}'
             WEBHOOK_HEADERS=('Content-Type'='application/json');
         GRANT USAGE ON INTEGRATION DEMO_NOTIFICATION_INTEGRATION TO ROLE ENGINEER;
         

1. All steps assume your working directory is the e2e_task_graph/ folder

   cd samples/ml/ml_jobs/e2e_task_graph
   

2. Configure your default Snowflake connection following the connection configuration guide

3. Set up your development environments using the setup_env.sh helper script. The script will create a new virtual environment (if needed), install all the necessary Python packages, and create the necessary resources in your default Snowflake account from step 2.

   bash scripts/setup_env.sh -r ENGINEER  # Change ENGINEER to your role name
   

   Modify -r ENGINEER to match the role name used in Snowflake Account Setup

4. Update the values in constants.py to match your Snowflake environment as configured in Snowflake Account Setup and any modifications made to step 3 above.

pipeline_local.py provides the core ML pipeline and can be executed locally for testing purposes.

The pipeline reads features from a Feature Store prepared in the previous setup step, generates train and test data splits, and runs model training using an ML Job. The pipeline concludes by evaluating the trained model and conditionally logging the trained model to Model Registry for downstream consumption.

Run the ML pipeline locally without task graph orchestration:

You can monitor the corresponding ML Job for model training via the Job UI in Snowsight.

pipeline_dag.py contains the Task Graph definition and can be used to trigger one-off executions or scheduled runs.

Deploy and immediately execute the task graph:

The script will:

Deploy the task graph with a recurring schedule:

Once your Task Graph is deployed and running, you can monitor and inspect your DAG executions through Snowflake's Task Graph UI in Snowsight.

Navigate to Monitoring > Task History to access the Task Graph interface:

Click into the Task Graph of interest to get a detailed view of the latest execution for that graph. In this case, click on DAG to inspect the Task Graph created in pipeline_dag.py

This visual interface makes it easy to:

pipeline_dag.py leverages several key Snowflake Task Graph features:

The train_model function uses the @remote decorator to run multi-node training on Snowpark Container Services:

The task graph includes branching logic that only promotes models meeting quality thresholds:

Successful models are automatically registered and promoted to production:

Congratulations! You've successfully built an end-to-end machine learning pipeline using Snowflake Task Graphs and ML Jobs. This production-ready solution demonstrates how to orchestrate the entire ML lifecycle within Snowflake's ecosystem, from data preparation to model deployment with conditional promotion logic.

YouTube video:

Web page:

Documentation:

More ML quickstart:

Happy building with Snowflake ML!