00_start_here.ipynb

{
 "metadata": {
  "kernelspec": {
   "display_name": "Streamlit Notebook",
   "name": "streamlit"
  }
 },
 "nbformat_minor": 5,
 "nbformat": 4,
 "cells": [
  {
   "cell_type": "markdown",
   "id": "623743e8-2cd7-47c6-99e7-100979384579",
   "metadata": {
    "name": "md_intro",
    "collapsed": false
   },
   "source": "# Snowflake Notebook Data Engineering\n\n* Author: Jeremiah Hansen\n* Last Updated: 6/11/2024\n\nWelcome to the beginning of the Quickstart! Please refer to [the official Snowflake Notebook Data Engineering Quickstart](https://quickstarts.snowflake.com/guide/data_engineering_with_notebooks/index.html?index=..%2F..index#0) for all the details including set up steps."
  },
  {
   "cell_type": "markdown",
   "id": "9e6273e5-bcf7-4492-92f6-cc161da082c6",
   "metadata": {
    "name": "md_step03",
    "collapsed": false
   },
   "source": "## Step 03 Setup Snowflake\n\nDuring this step we will create our demo environment. Update the SQL variables below with your GitHub username and Personal Access Token (PAT) as well as with your forked GitHub repository information.\n\n**Important**: Please make sure you have created the `dev` branch in your forked repository before continuing here. For instructions please see [Step 2 in the Quickstart](https://quickstarts.snowflake.com/guide/data_engineering_with_notebooks/index.html?index=..%2F..index#1)."
  },
  {
   "cell_type": "code",
   "id": "e898c514-831d-4aa7-9697-004994953950",
   "metadata": {
    "language": "sql",
    "name": "sql_step03_set_context"
   },
   "outputs": [],
   "source": "SET MY_USER = CURRENT_USER();\n\nSET GITHUB_SECRET_USERNAME = 'username';\nSET GITHUB_SECRET_PASSWORD = 'personal access token';\nSET GITHUB_URL_PREFIX = 'https://github.com/username';\nSET GITHUB_REPO_ORIGIN = 'https://github.com/username/sfguide-data-engineering-with-notebooks.git';",
   "execution_count": null
  },
  {
   "cell_type": "code",
   "id": "dc608c96-0957-47e1-8492-bc8d382925e3",
   "metadata": {
    "language": "sql",
    "name": "sql_step03_account_objects"
   },
   "outputs": [],
   "source": "-- ----------------------------------------------------------------------------\n-- Create the account level objects (ACCOUNTADMIN part)\n-- ----------------------------------------------------------------------------\n\nUSE ROLE ACCOUNTADMIN;\n\n-- Roles\nCREATE OR REPLACE ROLE DEMO_ROLE;\nGRANT ROLE DEMO_ROLE TO ROLE SYSADMIN;\nGRANT ROLE DEMO_ROLE TO USER IDENTIFIER($MY_USER);\n\nGRANT CREATE INTEGRATION ON ACCOUNT TO ROLE DEMO_ROLE;\nGRANT EXECUTE TASK ON ACCOUNT TO ROLE DEMO_ROLE;\nGRANT EXECUTE MANAGED TASK ON ACCOUNT TO ROLE DEMO_ROLE;\nGRANT MONITOR EXECUTION ON ACCOUNT TO ROLE DEMO_ROLE;\nGRANT IMPORTED PRIVILEGES ON DATABASE SNOWFLAKE TO ROLE DEMO_ROLE;\n\n-- Databases\nCREATE OR REPLACE DATABASE DEMO_DB;\nGRANT OWNERSHIP ON DATABASE DEMO_DB TO ROLE DEMO_ROLE;\n\n-- Warehouses\nCREATE OR REPLACE WAREHOUSE DEMO_WH WAREHOUSE_SIZE = XSMALL, AUTO_SUSPEND = 300, AUTO_RESUME= TRUE;\nGRANT OWNERSHIP ON WAREHOUSE DEMO_WH TO ROLE DEMO_ROLE;",
   "execution_count": null
  },
  {
   "cell_type": "code",
   "id": "a1e2ae2c-241b-4d8f-aa99-11a35f9833a4",
   "metadata": {
    "language": "sql",
    "name": "sql_step03_database_objects"
   },
   "outputs": [],
   "source": "-- ----------------------------------------------------------------------------\n-- Create the database level objects\n-- ----------------------------------------------------------------------------\nUSE ROLE DEMO_ROLE;\nUSE WAREHOUSE DEMO_WH;\nUSE DATABASE DEMO_DB;\n\n-- Schemas\nCREATE OR REPLACE SCHEMA INTEGRATIONS;\nCREATE OR REPLACE SCHEMA DEV_SCHEMA;\nCREATE OR REPLACE SCHEMA PROD_SCHEMA;\n\nUSE SCHEMA INTEGRATIONS;\n\n-- External Frostbyte objects\nCREATE OR REPLACE STAGE FROSTBYTE_RAW_STAGE\n    URL = 's3://sfquickstarts/data-engineering-with-snowpark-python/'\n;\n\n-- Secrets (schema level)\nCREATE OR REPLACE SECRET DEMO_GITHUB_SECRET\n  TYPE = password\n  USERNAME = $GITHUB_SECRET_USERNAME\n  PASSWORD = $GITHUB_SECRET_PASSWORD;\n\n-- API Integration (account level)\n-- This depends on the schema level secret!\nCREATE OR REPLACE API INTEGRATION DEMO_GITHUB_API_INTEGRATION\n  API_PROVIDER = GIT_HTTPS_API\n  API_ALLOWED_PREFIXES = ($GITHUB_URL_PREFIX)\n  ALLOWED_AUTHENTICATION_SECRETS = (DEMO_GITHUB_SECRET)\n  ENABLED = TRUE;\n\n-- Git Repository\nCREATE OR REPLACE GIT REPOSITORY DEMO_GIT_REPO\n  API_INTEGRATION = DEMO_GITHUB_API_INTEGRATION\n  GIT_CREDENTIALS = DEMO_GITHUB_SECRET\n  ORIGIN = $GITHUB_REPO_ORIGIN;",
   "execution_count": null
  },
  {
   "cell_type": "code",
   "id": "06f26add-547e-4d60-8897-d5ad79b3311d",
   "metadata": {
    "language": "sql",
    "name": "sql_step03_event_table"
   },
   "outputs": [],
   "source": "-- ----------------------------------------------------------------------------\n-- Create the event table\n-- ----------------------------------------------------------------------------\nUSE ROLE ACCOUNTADMIN;\n\nCREATE EVENT TABLE DEMO_DB.INTEGRATIONS.DEMO_EVENTS;\nGRANT SELECT ON EVENT TABLE DEMO_DB.INTEGRATIONS.DEMO_EVENTS TO ROLE DEMO_ROLE;\nGRANT INSERT ON EVENT TABLE DEMO_DB.INTEGRATIONS.DEMO_EVENTS TO ROLE DEMO_ROLE;\n\nALTER ACCOUNT SET EVENT_TABLE = DEMO_DB.INTEGRATIONS.DEMO_EVENTS;\nALTER DATABASE DEMO_DB SET LOG_LEVEL = INFO;",
   "execution_count": null
  },
  {
   "cell_type": "markdown",
   "id": "9531119f-76fc-4a2f-a635-a5a7526ac152",
   "metadata": {
    "name": "md_step04_deploy_dev_notebooks",
    "collapsed": false
   },
   "source": "## Step 04 Deploy to Dev\n\nFinally we will use `EXECUTE IMMEDIATE FROM <file>` along with Jinja templating to deploy the Dev version of our Notebooks. We will directly execute the SQL script `scripts/deploy_notebooks.sql` from our Git repository which has the SQL commands to deploy a Notebook from a Git repo.\n\nSee [EXECUTE IMMEDIATE FROM](https://docs.snowflake.com/en/sql-reference/sql/execute-immediate-from) for more details."
  },
  {
   "cell_type": "code",
   "id": "ad8676d0-7f82-4639-a5e2-29f7f9dca0f5",
   "metadata": {
    "language": "sql",
    "name": "sql_step04_deploy_dev_notebooks",
    "collapsed": false
   },
   "outputs": [],
   "source": "USE ROLE DEMO_ROLE;\nUSE WAREHOUSE DEMO_WH;\nUSE SCHEMA DEMO_DB.INTEGRATIONS;\n\nEXECUTE IMMEDIATE FROM @DEMO_GIT_REPO/branches/main/scripts/deploy_notebooks.sql\n    USING (env => 'DEV', branch => 'dev');",
   "execution_count": null
  },
  {
   "cell_type": "markdown",
   "id": "753bb327-95e4-4559-b7c7-f034607196c9",
   "metadata": {
    "name": "md_step05",
    "collapsed": false
   },
   "source": "## Step 05 Load Weather\n\nBut what about data that needs constant updating - like the WEATHER data? We would need to build a pipeline process to constantly update that data to keep it fresh.\n\nPerhaps a better way to get this external data would be to source it from a trusted data supplier. Let them manage the data, keeping it accurate and up to date.\n\nEnter the Snowflake Data Cloud...\n\nWeather Source is a leading provider of global weather and climate data and their OnPoint Product Suite provides businesses with the necessary weather and climate data to quickly generate meaningful and actionable insights for a wide range of use cases across industries. Let's connect to the \"Weather Source LLC: frostbyte\" feed from Weather Source in the Snowflake Data Marketplace by following these steps in Snowsight\n\n* In the left navigation bar click on \"Data Products\" and then \"Marketplace\"\n* Search: \"Weather Source LLC: frostbyte\" (and click on tile in results)\n* Click the blue \"Get\" button\n* Under \"Options\", adjust the Database name to read \"FROSTBYTE_WEATHERSOURCE\" (all capital letters)\n* Grant to \"HOL_ROLE\"\n\nThat's it... we don't have to do anything from here to keep this data updated. The provider will do that for us and data sharing means we are always seeing whatever they they have published."
  },
  {
   "cell_type": "code",
   "id": "04a850e3-44a4-4829-882e-84724f7e77d7",
   "metadata": {
    "language": "sql",
    "name": "sql_step05_create_share"
   },
   "outputs": [],
   "source": "/*---\n-- You can also do it via code if you know the account/share details...\nSET WEATHERSOURCE_ACCT_NAME = '*** PUT ACCOUNT NAME HERE AS PART OF DEMO SETUP ***';\nSET WEATHERSOURCE_SHARE_NAME = '*** PUT ACCOUNT SHARE HERE AS PART OF DEMO SETUP ***';\nSET WEATHERSOURCE_SHARE = $WEATHERSOURCE_ACCT_NAME || '.' || $WEATHERSOURCE_SHARE_NAME;\n\nCREATE OR REPLACE DATABASE FROSTBYTE_WEATHERSOURCE\n  FROM SHARE IDENTIFIER($WEATHERSOURCE_SHARE);\n\nGRANT IMPORTED PRIVILEGES ON DATABASE FROSTBYTE_WEATHERSOURCE TO ROLE HOL_ROLE;\n---*/",
   "execution_count": null
  },
  {
   "cell_type": "code",
   "id": "7e2762d1-fe91-4a7c-b89a-56e1baf0001c",
   "metadata": {
    "language": "sql",
    "name": "sql_step05_test_share"
   },
   "outputs": [],
   "source": "-- Let's look at the data - same 3-part naming convention as any other table\nSELECT * FROM FROSTBYTE_WEATHERSOURCE.ONPOINT_ID.POSTAL_CODES LIMIT 100;",
   "execution_count": null
  },
  {
   "cell_type": "markdown",
   "id": "d3d93974-9a75-46c2-876f-95b6e1562f75",
   "metadata": {
    "name": "md_step06",
    "collapsed": false
   },
   "source": "## Step 06 Load Excel Files\n\nPlease follow the instructions in [Step 6 of the Quickstart](https://quickstarts.snowflake.com/guide/data_engineering_with_notebooks/index.html?index=..%2F..index#5) to open and run the `DEV_06_load_excel_files` Notebook. That Notebook will define the pipeline used to load data into the `LOCATION` and `ORDER_DETAIL` tables from the staged Excel files."
  },
  {
   "cell_type": "markdown",
   "id": "b5587283-a264-444d-b9ec-55b4d926a5c7",
   "metadata": {
    "name": "md_step07",
    "collapsed": false
   },
   "source": "## Step 07 Load Daily City Metrics\n\nPlease follow the instructions in [Step 7 of the Quickstart](https://quickstarts.snowflake.com/guide/data_engineering_with_notebooks/index.html?index=..%2F..index#6) to open and run the `DEV_07_load_daily_city_metrics` Notebook. That Notebook will define the pipeline used to create the `DAILY_CITY_METRICS` table."
  },
  {
   "cell_type": "code",
   "id": "63bfff6b-067e-4f24-8424-19d0231c0ee1",
   "metadata": {
    "language": "sql",
    "name": "sql_step07_logs"
   },
   "outputs": [],
   "source": "USE ROLE DEMO_ROLE;\nUSE WAREHOUSE DEMO_WH;\nUSE SCHEMA DEMO_DB.INTEGRATIONS;\n\nSELECT TOP 100\n  RECORD['severity_text'] AS SEVERITY,\n  VALUE AS MESSAGE\nFROM\n  DEMO_DB.INTEGRATIONS.DEMO_EVENTS\nWHERE 1 = 1\n  AND SCOPE['name'] = 'demo_logger'\n  AND RECORD_TYPE = 'LOG';",
   "execution_count": null
  },
  {
   "cell_type": "markdown",
   "id": "c873a1db-1fbe-4a44-b02a-03e1b2084cb2",
   "metadata": {
    "name": "md_step08",
    "collapsed": false
   },
   "source": "## Step 08 Orchestrate Pipelines\n\nIn this step we will create a DAG (or Directed Acyclic Graph) of Tasks using the new [Snowflake Python Management API](https://docs.snowflake.com/en/developer-guide/snowflake-python-api/snowflake-python-overview). The Task DAG API builds upon the Python Management API to provide advanced Task management capabilities. For more details see [Managing Snowflake tasks and task graphs with Python](https://docs.snowflake.com/en/developer-guide/snowflake-python-api/snowflake-python-managing-tasks).\n\nThis code is also available in the `scripts/deploy_task_dag.py` script which could be used to automate the Task DAG deployment."
  },
  {
   "cell_type": "code",
   "id": "bdac9ad2-2858-4fb7-b3a2-6394db5b0b4c",
   "metadata": {
    "language": "python",
    "name": "py_step08_imports"
   },
   "outputs": [],
   "source": "# Import python packages\nfrom snowflake.core import Root\n\n# We can also use Snowpark for our analyses!\nfrom snowflake.snowpark.context import get_active_session\nsession = get_active_session()\n\nsession.use_role(\"DEMO_ROLE\")\nsession.use_warehouse(\"DEMO_WH\")",
   "execution_count": null
  },
  {
   "cell_type": "code",
   "id": "cb2bbc8e-c525-4cd0-b147-a832a9060c47",
   "metadata": {
    "language": "python",
    "name": "py_step08_set_env"
   },
   "outputs": [],
   "source": "database_name = \"DEMO_DB\"\nschema_name = \"DEV_SCHEMA\"\n#schema_name = \"PROD_SCHEMA\"\nenv = 'PROD' if schema_name == 'PROD_SCHEMA' else 'DEV'\n\nsession.use_schema(f\"{database_name}.{schema_name}\")",
   "execution_count": null
  },
  {
   "cell_type": "code",
   "id": "0c030976-5888-4d9f-a329-3248b25abd78",
   "metadata": {
    "language": "python",
    "name": "py_step08_create_dag"
   },
   "outputs": [],
   "source": "from snowflake.core.task.dagv1 import DAGOperation, DAG, DAGTask\nfrom datetime import timedelta\n\n# Create the tasks using the DAG API\nwarehouse_name = \"DEMO_WH\"\ndag_name = \"DEMO_DAG\"\n\napi_root = Root(session)\nschema = api_root.databases[database_name].schemas[schema_name]\ndag_op = DAGOperation(schema)\n\n# Define the DAG\nwith DAG(dag_name, schedule=timedelta(days=1), warehouse=warehouse_name) as dag:\n    dag_task1 = DAGTask(\"LOAD_EXCEL_FILES_TASK\", definition=f'''EXECUTE NOTEBOOK \"{database_name}\".\"{schema_name}\".\"{env}_06_load_excel_files\"()''', warehouse=warehouse_name)\n    dag_task2 = DAGTask(\"LOAD_DAILY_CITY_METRICS\", definition=f'''EXECUTE NOTEBOOK \"{database_name}\".\"{schema_name}\".\"{env}_07_load_daily_city_metrics\"()''', warehouse=warehouse_name)\n\n    # Define the dependencies between the tasks\n    dag_task1 >> dag_task2 # dag_task1 is a predecessor of dag_task2\n\n# Create the DAG in Snowflake\ndag_op.deploy(dag, mode=\"orreplace\")",
   "execution_count": null
  },
  {
   "cell_type": "code",
   "id": "f560c909-5523-4a12-ad21-0b9044bdaff6",
   "metadata": {
    "language": "python",
    "name": "py_step08_run_dag"
   },
   "outputs": [],
   "source": "dagiter = dag_op.iter_dags(like='demo_dag%')\nfor dag_name in dagiter:\n    print(dag_name)\n\n#dag_op.run(dag)",
   "execution_count": null
  },
  {
   "cell_type": "markdown",
   "id": "8ad46ffc-1137-43dc-add8-7fc02914bbaa",
   "metadata": {
    "name": "md_step09",
    "collapsed": false
   },
   "source": "## Step 09 Deploy to Production\n\nSteps\n1. Make a small change to a notebook and commit it to the dev branch\n1. Go into GitHub and create a PR and Merge to main branch\n1. Review GitHub Actions workflow definition and run results\n1. See new \"PROD_\" versions of the Notebooks\n1. Deploy the production version of the task DAG\n1. Run production version of the task DAG and see new tables created!"
  },
  {
   "cell_type": "markdown",
   "id": "ba497c01-0988-4c07-af66-79ee2918cffa",
   "metadata": {
    "name": "md_step10",
    "collapsed": false
   },
   "source": "## Step 10 Teardown\n\nFinally, we will tear down our demo environment."
  },
  {
   "cell_type": "code",
   "id": "f47ca116-4585-4668-bb72-cf74b0e7b587",
   "metadata": {
    "language": "sql",
    "name": "sql_step10"
   },
   "outputs": [],
   "source": "USE ROLE ACCOUNTADMIN;\n\nDROP API INTEGRATION DEMO_GITHUB_API_INTEGRATION;\nDROP DATABASE DEMO_DB;\nDROP WAREHOUSE DEMO_WH;\nDROP ROLE DEMO_ROLE;\n\n-- Drop the weather share\nDROP DATABASE FROSTBYTE_WEATHERSOURCE;\n\n-- Remove the \"dev\" branch in your repo",
   "execution_count": null
  }
 ]
}