Data Engineering Pipelines with Snowpark Python/Data Engineering Pipelines with Snowpark Python.ipynb

{
  "cells": [
    {
      "cell_type": "markdown",
      "id": "152b9c5c-1d6a-4792-9f21-cb198d9554cb",
      "metadata": {
        "collapsed": false,
        "jupyter": {
          "outputs_hidden": false
        },
        "name": "cell1"
      },
      "source": [
        "# Data Engineering Pipelines with Snowpark Python\n",
        "\n",
        "\n",
        "\"Data engineers are focused primarily on building and maintaining data pipelines that transport data through different steps and put it into a usable state ... The data engineering process encompasses the overall effort required to create **data pipelines** that automate the transfer of data from place to place and transform that data into a specific format for a certain type of analysis. In that sense, data engineering isn\u2019t something you do once. It\u2019s an ongoing practice that involves collecting, preparing, transforming, and delivering data. A data pipeline helps automate these tasks so they can be reliably repeated. It\u2019s a practice more than a specific technology.\" (From Cloud Data Engineering for Dummies, Snowflake Special Edition)\n",
        "\n",
        "Are you interested in unleashing the power of Snowpark Python to build data engineering pipelines? Well then, this Quickstart is for you! The focus here will be on building data engineering pipelines with Python, and not on data science. For examples of doing data science with Snowpark Python please check out our [Machine Learning with Snowpark Python: - Credit Card Approval Prediction](https://quickstarts.snowflake.com/guide/getting_started_snowpark_machine_learning/index.html?index=..%2F..index#0) Quickstart.\n",
        "\n",
        "[This Quickstart](https://quickstarts.snowflake.com/guide/data_engineering_pipelines_with_snowpark_python/index.html) will cover a lot of ground, and by the end you will have built a robust data engineering pipeline using Snowpark Python stored procedures. That pipeline will process data incrementally, be orchestrated with Snowflake tasks, and be deployed via a CI/CD pipeline. You'll also learn how to use Snowflake's new developer CLI tool and Visual Studio Code extension! Here's a quick visual overview:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "id": "63cfc92d-977c-4019-a64e-a297b49b356a",
      "metadata": {
        "codeCollapsed": false,
        "language": "python",
        "name": "cell2"
      },
      "outputs": [],
      "source": [
        "import streamlit as st\n",
        "st.image(\"https://raw.githubusercontent.com/Snowflake-Labs/sfguide-data-engineering-with-snowpark-python/main/images/demo_overview.png\",width=800)"
      ]
    },
    {
      "cell_type": "markdown",
      "id": "5a796a19-195f-488b-8a54-240de881582c",
      "metadata": {
        "collapsed": false,
        "jupyter": {
          "outputs_hidden": false
        },
        "name": "cell3"
      },
      "source": [
        "### Prerequisites\n",
        "* Familiarity with Python\n",
        "* Familiarity with the DataFrame API\n",
        "* Familiarity with Snowflake\n",
        "\n",
        "### What You\u2019ll Learn\n",
        "You will learn about the following Snowflake features during this Quickstart:\n",
        "\n",
        "* Snowflake's Table Format\n",
        "* Data ingestion with COPY\n",
        "* Schema inference\n",
        "* Data sharing/marketplace (instead of ETL)\n",
        "* Streams for incremental processing (CDC)\n",
        "* Streams on views\n",
        "* Python UDFs (with third-party packages)\n",
        "* Python Stored Procedures\n",
        "* Snowpark DataFrame API\n",
        "* Snowpark Python programmability\n",
        "* Warehouse elasticity (dynamic scaling)\n",
        "* SnowCLI (PuPr)\n",
        "* Tasks (with Stream triggers)\n",
        "* Task Observability\n",
        "\n",
        "### What You\u2019ll Need\n",
        "You will need the following things before beginning:\n",
        "\n",
        "* Snowflake account\n",
        "    * **A Snowflake Account**\n",
        "    * **A Snowflake user created with ACCOUNTADMIN permissions**. This user will be used to get things setup in Snowflake.\n",
        "    * **Anaconda Terms & Conditions accepted**. See Getting Started section in [Third-Party Packages](https://docs.snowflake.com/en/developer-guide/udf/python/udf-python-packages.html#getting-started).\n",
        "\n",
        "### What You\u2019ll Build\n",
        "During this Quickstart you will accomplish the following things:\n",
        "\n",
        "* Load Parquet data to Snowflake using schema inference\n",
        "* Setup access to Snowflake Marketplace data\n",
        "* Create a Python UDF to convert temperature\n",
        "* Create a data engineering pipeline with Python stored procedures to incrementally process data\n",
        "* Orchestrate the pipelines with tasks\n",
        "* Monitor the pipelines with Snowsight\n",
        "* Deploy the Snowpark Python stored procedures via a CI/CD pipeline\n",
        "\n",
        "\n",
        "## Setup Snowflake\n",
        "\n",
        "You can run SQL queries against Snowflake in many different ways (through the Snowsight UI, SnowSQL, etc.) but for this Quickstart we'll be using the Snowflake Notebooks.\n",
        "\n",
        "### Run the Script\n",
        "To set up all the objects we'll need in Snowflake for this Quickstart you'll need to run following steps."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "id": "7551d60d-f4e1-44cb-be72-ce26d4814c50",
      "metadata": {
        "codeCollapsed": false,
        "collapsed": false,
        "jupyter": {
          "outputs_hidden": false
        },
        "language": "sql",
        "name": "cell4"
      },
      "outputs": [],
      "source": [
        "-- ----------------------------------------------------------------------------\n",
        "-- Step #1: Accept Anaconda Terms & Conditions\n",
        "-- ----------------------------------------------------------------------------\n",
        "\n",
        "-- See Getting Started section in Third-Party Packages (https://docs.snowflake.com/en/developer-guide/udf/python/udf-python-packages.html#getting-started)\n",
        "\n",
        "\n",
        "-- ----------------------------------------------------------------------------\n",
        "-- Step #2: Create the account level objects\n",
        "-- ----------------------------------------------------------------------------\n",
        "USE ROLE ACCOUNTADMIN;\n",
        "\n",
        "-- Roles\n",
        "SET MY_USER = CURRENT_USER(); \n",
        "CREATE OR REPLACE ROLE HOL_ROLE;\n",
        "GRANT ROLE HOL_ROLE TO ROLE SYSADMIN;\n",
        "GRANT ROLE HOL_ROLE TO USER IDENTIFIER($MY_USER);\n",
        "\n",
        "CREATE STAGE IF NOT EXISTS SCRIPTS \n",
        "\tDIRECTORY = ( ENABLE = true );\n",
        "\n",
        "GRANT EXECUTE TASK ON ACCOUNT TO ROLE HOL_ROLE;\n",
        "GRANT MONITOR EXECUTION ON ACCOUNT TO ROLE HOL_ROLE;\n",
        "GRANT IMPORTED PRIVILEGES ON DATABASE SNOWFLAKE TO ROLE HOL_ROLE;\n",
        "\n",
        "-- Databases\n",
        "CREATE OR REPLACE DATABASE NB_HOL_DB;\n",
        "GRANT OWNERSHIP ON DATABASE NB_HOL_DB TO ROLE HOL_ROLE;\n",
        "\n",
        "-- Warehouses\n",
        "CREATE OR REPLACE WAREHOUSE HOL_WH WAREHOUSE_SIZE = XSMALL, AUTO_SUSPEND = 300, AUTO_RESUME= TRUE;\n",
        "GRANT OWNERSHIP ON WAREHOUSE HOL_WH TO ROLE HOL_ROLE;"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "id": "b706ef72-370b-41c4-a46f-bb3eb8f58f45",
      "metadata": {
        "codeCollapsed": false,
        "collapsed": false,
        "language": "sql",
        "name": "cell5"
      },
      "outputs": [],
      "source": [
        "-- ----------------------------------------------------------------------------\n",
        "-- Step #3: Create the database level objects\n",
        "-- ----------------------------------------------------------------------------\n",
        "USE ROLE HOL_ROLE;\n",
        "USE WAREHOUSE HOL_WH;\n",
        "USE DATABASE NB_HOL_DB;\n",
        "\n",
        "-- Schemas\n",
        "CREATE OR REPLACE SCHEMA EXTERNAL;\n",
        "CREATE OR REPLACE SCHEMA RAW_POS;\n",
        "CREATE OR REPLACE SCHEMA RAW_CUSTOMER;\n",
        "CREATE OR REPLACE SCHEMA HARMONIZED;\n",
        "CREATE OR REPLACE SCHEMA ANALYTICS;\n",
        "\n",
        "-- External Frostbyte objects\n",
        "USE SCHEMA EXTERNAL;\n",
        "CREATE OR REPLACE FILE FORMAT PARQUET_FORMAT\n",
        "    TYPE = PARQUET\n",
        "    COMPRESSION = SNAPPY;\n",
        "    \n",
        "CREATE OR REPLACE STAGE FROSTBYTE_RAW_STAGE\n",
        "    URL = 's3://sfquickstarts/data-engineering-with-snowpark-python/';\n",
        "\n",
        "-- ANALYTICS objects\n",
        "USE SCHEMA ANALYTICS;\n",
        "\n",
        "CREATE OR REPLACE FUNCTION ANALYTICS.INCH_TO_MILLIMETER_UDF(INCH NUMBER(35,4))\n",
        "RETURNS NUMBER(35,4)\n",
        "    AS\n",
        "$$\n",
        "    inch * 25.4\n",
        "$$;"
      ]
    },
    {
      "cell_type": "markdown",
      "id": "a57b699e-0912-4f0c-b23b-7a3ddf42a93a",
      "metadata": {
        "collapsed": false,
        "jupyter": {
          "outputs_hidden": false
        },
        "name": "cell6"
      },
      "source": [
        "### Load Raw \n",
        "\n",
        "During this step we will be loading the raw Tasty Bytes POS and Customer loyalty data from raw Parquet files in `s3://sfquickstarts/data-engineering-with-snowpark-python/` to our `RAW_POS` and `RAW_CUSTOMER` schemas in Snowflake. And you are going to be orchestrating this process from your laptop in Python using the Snowpark Python API. To put this in context, we are on step **#2** in our data flow overview:\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "id": "a55a0d00-69d2-4aae-9c68-755d5c30fb4c",
      "metadata": {
        "codeCollapsed": false,
        "language": "python",
        "name": "cell7"
      },
      "outputs": [],
      "source": [
        "import streamlit as st\n",
        "st.image(\"https://raw.githubusercontent.com/Snowflake-Labs/sfguide-data-engineering-with-snowpark-python/main/images/demo_overview.png\",width=800)"
      ]
    },
    {
      "cell_type": "markdown",
      "id": "edba6e43-badd-416f-9a62-7cf77cf248b1",
      "metadata": {
        "name": "cell8"
      },
      "source": [
        "To load the raw data, execute the following:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "id": "bdbbe721-66fc-4031-b3fe-6f195ee5baa7",
      "metadata": {
        "codeCollapsed": false,
        "collapsed": false,
        "jupyter": {
          "outputs_hidden": false
        },
        "language": "python",
        "name": "cell9"
      },
      "outputs": [],
      "source": [
        "import time\n",
        "from snowflake.snowpark import Session\n",
        "#import snowflake.snowpark.types as T\n",
        "#import snowflake.snowpark.functions as F\n",
        "\n",
        "\n",
        "POS_TABLES = ['country', 'franchise', 'location', 'menu', 'truck', 'order_header', 'order_detail']\n",
        "CUSTOMER_TABLES = ['customer_loyalty']\n",
        "TABLE_DICT = {\n",
        "    \"pos\": {\"schema\": \"RAW_POS\", \"tables\": POS_TABLES},\n",
        "    \"customer\": {\"schema\": \"RAW_CUSTOMER\", \"tables\": CUSTOMER_TABLES}\n",
        "}\n",
        "\n",
        "# SNOWFLAKE ADVANTAGE: Schema detection\n",
        "# SNOWFLAKE ADVANTAGE: Data ingestion with COPY\n",
        "# SNOWFLAKE ADVANTAGE: Snowflake Tables (not file-based)\n",
        "\n",
        "def load_raw_table(session, tname=None, s3dir=None, year=None, schema=None):\n",
        "    session.use_schema(schema)\n",
        "    if year is None:\n",
        "        location = \"@external.frostbyte_raw_stage/{}/{}\".format(s3dir, tname)\n",
        "    else:\n",
        "        print('\\tLoading year {}'.format(year)) \n",
        "        location = \"@external.frostbyte_raw_stage/{}/{}/year={}\".format(s3dir, tname, year)\n",
        "    \n",
        "    # we can infer schema using the parquet read option\n",
        "    df = session.read.option(\"compression\", \"snappy\") \\\n",
        "                            .parquet(location)\n",
        "    df.copy_into_table(\"{}\".format(tname))\n",
        "\n",
        "# SNOWFLAKE ADVANTAGE: Warehouse elasticity (dynamic scaling)\n",
        "\n",
        "def load_all_raw_tables(session):\n",
        "    _ = session.sql(\"ALTER WAREHOUSE HOL_WH SET WAREHOUSE_SIZE = XLARGE WAIT_FOR_COMPLETION = TRUE\").collect()\n",
        "\n",
        "    for s3dir, data in TABLE_DICT.items():\n",
        "        tnames = data['tables']\n",
        "        schema = data['schema']\n",
        "        for tname in tnames:\n",
        "            print(\"Loading {}\".format(tname))\n",
        "            # Only load the first 3 years of data for the order tables at this point\n",
        "            # We will load the 2022 data later in the lab\n",
        "            if tname in ['order_header', 'order_detail']:\n",
        "                # For testing, use only records from 2019. Uncomment to add more years of data.\n",
        "                for year in ['2019']:#, '2020', '2021']: \n",
        "                    load_raw_table(session, tname=tname, s3dir=s3dir, year=year, schema=schema)\n",
        "            else:\n",
        "                load_raw_table(session, tname=tname, s3dir=s3dir, schema=schema)\n",
        "\n",
        "    _ = session.sql(\"ALTER WAREHOUSE HOL_WH SET WAREHOUSE_SIZE = XSMALL\").collect()\n",
        "\n",
        "def validate_raw_tables(session):\n",
        "    # check column names from the inferred schema\n",
        "    for tname in POS_TABLES:\n",
        "        print('{}: \\n\\t{}\\n'.format(tname, session.table('RAW_POS.{}'.format(tname)).columns))\n",
        "\n",
        "    for tname in CUSTOMER_TABLES:\n",
        "        print('{}: \\n\\t{}\\n'.format(tname, session.table('RAW_CUSTOMER.{}'.format(tname)).columns))"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "id": "f6659e70-913c-4978-8b2e-d154843328d7",
      "metadata": {
        "codeCollapsed": false,
        "collapsed": false,
        "jupyter": {
          "outputs_hidden": false
        },
        "language": "python",
        "name": "cell10"
      },
      "outputs": [],
      "source": [
        "# Add the utils package to our path and import the snowpark_utils function\n",
        "import os, sys\n",
        "current_dir = os.getcwd()\n",
        "parent_dir = os.path.dirname(current_dir)\n",
        "sys.path.append(parent_dir)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "id": "52cf12e2-9257-47b0-8c55-c7d87cce6267",
      "metadata": {
        "codeCollapsed": false,
        "collapsed": false,
        "jupyter": {
          "outputs_hidden": false
        },
        "language": "python",
        "name": "cell11"
      },
      "outputs": [],
      "source": [
        "from snowflake.snowpark.context import get_active_session\n",
        "session = get_active_session()"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "id": "f624c8ad-f94b-4cf2-b06d-76b32b39a507",
      "metadata": {
        "codeCollapsed": false,
        "collapsed": false,
        "jupyter": {
          "outputs_hidden": false
        },
        "language": "python",
        "name": "cell12"
      },
      "outputs": [],
      "source": [
        "load_all_raw_tables(session)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "id": "62214d74-b02d-4868-8ad2-e6967d581ab0",
      "metadata": {
        "codeCollapsed": false,
        "collapsed": false,
        "jupyter": {
          "outputs_hidden": false
        },
        "language": "python",
        "name": "cell13"
      },
      "outputs": [],
      "source": [
        "validate_raw_tables(session)"
      ]
    },
    {
      "cell_type": "markdown",
      "id": "9c058d54-6ad1-429b-813b-d8309a433f7a",
      "metadata": {
        "collapsed": false,
        "name": "cell14"
      },
      "source": [
        "### Viewing What Happened in Snowflake\n",
        "The [Query History](https://docs.snowflake.com/en/user-guide/ui-snowsight-activity.html#query-history) in Snowflake is a very power feature, that logs every query run against your Snowflake account, no matter which tool or process initiated it. And this is especially helpful when working with client tools and APIs.\n",
        "\n",
        "The Python script you just ran did a small amount of work locally, basically just orchestrating the process by looping through each table and issuing the command to Snowflake to load the data. But all of the heavy lifting ran inside Snowflake! This push-down is a hallmark of the Snowpark API and allows you to leverage the scalability and compute power of Snowflake!\n",
        "\n",
        "Log in to your Snowflake account and take a quick look at the SQL that was generated by the Snowpark API. This will help you better understand what the API is doing and will help you debug any issues you may run into.\n",
        "\n",
        "\n",
        "### Schema Inference\n",
        "One very helpful feature in Snowflake is the ability to infer the schema of files in a stage that you wish to work with. This is accomplished in SQL with the [`INFER_SCHEMA()`](https://docs.snowflake.com/en/sql-reference/functions/infer_schema.html) function. The Snowpark Python API does this for you automatically when you call the `session.read()` method. Here is the code snippet:\n",
        "\n",
        "```python\n",
        "    # we can infer schema using the parquet read option\n",
        "    df = session.read.option(\"compression\", \"snappy\") \\\n",
        "                            .parquet(location)\n",
        "```\n",
        "\n",
        "### Data Ingestion with COPY\n",
        "In order to load the data into a Snowflake table we will use the `copy_into_table()` method on a DataFrame. This method will create the target table in Snowflake using the inferred schema (if it doesn't exist), and then call the highly optimized Snowflake [`COPY INTO &lt;table&gt;` Command](https://docs.snowflake.com/en/sql-reference/sql/copy-into-table.html). Here is the code snippet:\n",
        "\n",
        "```python\n",
        "    df.copy_into_table(\"{}\".format(tname))\n",
        "```\n",
        "\n",
        "### Snowflake's Table Format\n",
        "One of the major advantages of Snowflake is being able to eliminate the need to manage a file-based data lake. And Snowflake was designed for this purpose from the beginning. In the step we are loading the raw data into a structured Snowflake managed table. But Snowflake tables can natively support structured and semi-structured data, and are stored in Snowflake's mature cloud table format (which predates Hudi, Delta or Iceberg).\n",
        "\n",
        "Once loaded into Snowflake the data will be securely stored and managed, without the need to worry about securing and managing raw files. Additionally the data, whether raw or structured, can be transformed and queried in Snowflake using SQL or the language of your choice, without needing to manage separate compute services like Spark.\n",
        "\n",
        "This is a huge advantage for Snowflake customers.\n",
        "\n",
        "\n",
        "### Warehouse Elasticity (Dynamic Scaling)\n",
        "With Snowflake there is only one type of user defined compute cluster, the [Virtual Warehouse](https://docs.snowflake.com/en/user-guide/warehouses.html), regardless of the language you use to process that data (SQL, Python, Java, Scala, Javascript, etc.). This makes working with data much simpler in Snowflake. And governance of the data is completely separated from the compute cluster, in other words there is no way to get around Snowflake governance regardless of the warehouse settings or language being used.\n",
        "\n",
        "And these virtual warehouses can be dynamically scaled, in under a second for most sized warehouses! This means that in your code you can dynamically resize the compute environment to increase the amount of capacity to run a section of code in a fraction of the time, and then dynamically resized again to reduce the amount of capacity. And because of our per-second billing (with a sixty second minimum) you won't pay any extra to run that section of code in a fraction of the time!\n",
        "\n",
        "Let's see how easy that is done. Here is the code snippet:\n",
        "\n",
        "```python\n",
        "_ = session.sql(\"ALTER WAREHOUSE HOL_WH SET WAREHOUSE_SIZE = XLARGE WAIT_FOR_COMPLETION = TRUE\").collect()\n",
        "\n",
        "# Some data processing code\n",
        "\n",
        "_ = session.sql(\"ALTER WAREHOUSE HOL_WH SET WAREHOUSE_SIZE = XSMALL\").collect()\n",
        "```\n",
        "\n",
        "Please also note that we included the `WAIT_FOR_COMPLETION` parameter in the first `ALTER WAREHOUSE` statement. Setting this parameter to `TRUE` will block the return of the `ALTER WAREHOUSE` command until the resize has finished provisioning all its compute resources. This way we make sure that the full cluster is available before processing any data with it.\n",
        "\n",
        "We will use this pattern a few more times during this Quickstart, so it's important to understand.\n"
      ]
    },
    {
      "cell_type": "markdown",
      "id": "64122344-5484-430d-907a-2ba3702d7172",
      "metadata": {
        "collapsed": false,
        "jupyter": {
          "outputs_hidden": false
        },
        "name": "cell15"
      },
      "source": [
        "## Load Weather\n",
        "\n",
        "\n",
        "During this step we will be \"loading\" the raw weather data to Snowflake. But \"loading\" is the really the wrong word here. Because we're using Snowflake's unique data sharing capability we don't actually need to copy the data to our Snowflake account with a custom ETL process. Instead we can directly access the weather data shared by Weather Source in the Snowflake Data Marketplace. To put this in context, we are on step **#3** in our data flow overview. \n",
        "\n",
        "\n",
        "### Snowflake Data Marketplace\n",
        "Weather 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:\n",
        "\n",
        "* Login to Snowsight\n",
        "* Click on the `Marketplace` link in the left navigation bar\n",
        "* Enter \"Weather Source LLC: frostbyte\" in the search box and click return\n",
        "* Click on the \"Weather Source LLC: frostbyte\" listing tile\n",
        "* Click the blue \"Get\" button\n",
        "    * Expand the \"Options\" dialog\n",
        "    * Change the \"Database name\" to read \"FROSTBYTE_WEATHERSOURCE\" (all capital letters)\n",
        "    * Select the \"HOL_ROLE\" role to have access to the new database\n",
        "* Click on the blue \"Get\" button\n",
        "\n",
        "That'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 have published. How amazing is that? Just think of all the things you didn't have do here to get access to an always up-to-date, third-party dataset!"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "id": "46b8555e-4c29-4d5c-8193-d182c822baf4",
      "metadata": {
        "codeCollapsed": false,
        "collapsed": false,
        "jupyter": {
          "outputs_hidden": false
        },
        "language": "sql",
        "name": "cell16"
      },
      "outputs": [],
      "source": [
        "-- Make sure you grant the table privileges to HOL_ROLE for this to run\n",
        "USE ROLE ACCOUNTADMIN;\n",
        "GRANT IMPORTED PRIVILEGES ON DATABASE IDENTIFIER('\"FROSTBYTE_WEATHERSOURCE\"') TO ROLE IDENTIFIER('\"HOL_ROLE\"');\n",
        "USE ROLE HOL_ROLE;\n",
        "-- Preview Weathersource data\n",
        "SELECT * FROM FROSTBYTE_WEATHERSOURCE.ONPOINT_ID.POSTAL_CODES LIMIT 100; "
      ]
    },
    {
      "cell_type": "markdown",
      "id": "9192a8d0-66fb-4a73-9883-531d25607a9c",
      "metadata": {
        "collapsed": false,
        "jupyter": {
          "outputs_hidden": false
        },
        "name": "cell17"
      },
      "source": [
        "## Create POS View\n",
        "\n",
        "During this step we will be creating a view to simplify the raw POS schema by joining together 6 different tables and picking only the columns we need. But what's really cool is that we're going to define that view with the Snowpark DataFrame API! Then we're going to create a Snowflake stream on that view so that we can incrementally process changes to any of the POS tables. To put this in context, we are on step **#4** in our data flow overview.\n",
        "\n",
        "### Run the Script\n",
        "To create the view and stream, execute the following code."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "id": "8235cfa2-7694-4d77-bbf2-7d3c71c45ab1",
      "metadata": {
        "codeCollapsed": false,
        "language": "python",
        "name": "cell18"
      },
      "outputs": [],
      "source": [
        "# SNOWFLAKE ADVANTAGE: Snowpark DataFrame API\n",
        "# SNOWFLAKE ADVANTAGE: Streams for incremental processing (CDC)\n",
        "# SNOWFLAKE ADVANTAGE: Streams on views\n",
        "\n",
        "\n",
        "from snowflake.snowpark import Session\n",
        "#import snowflake.snowpark.types as T\n",
        "import snowflake.snowpark.functions as F\n",
        "\n",
        "\n",
        "def create_pos_view(session):\n",
        "    session.use_schema('HARMONIZED')\n",
        "    order_detail = session.table(\"RAW_POS.ORDER_DETAIL\").select(F.col(\"ORDER_DETAIL_ID\"), \\\n",
        "                                                                F.col(\"LINE_NUMBER\"), \\\n",
        "                                                                F.col(\"MENU_ITEM_ID\"), \\\n",
        "                                                                F.col(\"QUANTITY\"), \\\n",
        "                                                                F.col(\"UNIT_PRICE\"), \\\n",
        "                                                                F.col(\"PRICE\"), \\\n",
        "                                                                F.col(\"ORDER_ID\"))\n",
        "    order_header = session.table(\"RAW_POS.ORDER_HEADER\").select(F.col(\"ORDER_ID\"), \\\n",
        "                                                                F.col(\"TRUCK_ID\"), \\\n",
        "                                                                F.col(\"ORDER_TS\"), \\\n",
        "                                                                F.to_date(F.col(\"ORDER_TS\")).alias(\"ORDER_TS_DATE\"), \\\n",
        "                                                                F.col(\"ORDER_AMOUNT\"), \\\n",
        "                                                                F.col(\"ORDER_TAX_AMOUNT\"), \\\n",
        "                                                                F.col(\"ORDER_DISCOUNT_AMOUNT\"), \\\n",
        "                                                                F.col(\"LOCATION_ID\"), \\\n",
        "                                                                F.col(\"ORDER_TOTAL\"))\n",
        "    truck = session.table(\"RAW_POS.TRUCK\").select(F.col(\"TRUCK_ID\"), \\\n",
        "                                                F.col(\"PRIMARY_CITY\"), \\\n",
        "                                                F.col(\"REGION\"), \\\n",
        "                                                F.col(\"COUNTRY\"), \\\n",
        "                                                F.col(\"FRANCHISE_FLAG\"), \\\n",
        "                                                F.col(\"FRANCHISE_ID\"))\n",
        "    menu = session.table(\"RAW_POS.MENU\").select(F.col(\"MENU_ITEM_ID\"), \\\n",
        "                                                F.col(\"TRUCK_BRAND_NAME\"), \\\n",
        "                                                F.col(\"MENU_TYPE\"), \\\n",
        "                                                F.col(\"MENU_ITEM_NAME\"))\n",
        "    franchise = session.table(\"RAW_POS.FRANCHISE\").select(F.col(\"FRANCHISE_ID\"), \\\n",
        "                                                        F.col(\"FIRST_NAME\").alias(\"FRANCHISEE_FIRST_NAME\"), \\\n",
        "                                                        F.col(\"LAST_NAME\").alias(\"FRANCHISEE_LAST_NAME\"))\n",
        "    location = session.table(\"RAW_POS.LOCATION\").select(F.col(\"LOCATION_ID\"))\n",
        "\n",
        "    \n",
        "    '''\n",
        "    We can do this one of two ways: either select before the join so it is more explicit, or just join on the full tables.\n",
        "    The end result is the same, it's mostly a readibility question.\n",
        "    '''\n",
        "    # order_detail = session.table(\"RAW_POS.ORDER_DETAIL\")\n",
        "    # order_header = session.table(\"RAW_POS.ORDER_HEADER\")\n",
        "    # truck = session.table(\"RAW_POS.TRUCK\")\n",
        "    # menu = session.table(\"RAW_POS.MENU\")\n",
        "    # franchise = session.table(\"RAW_POS.FRANCHISE\")\n",
        "    # location = session.table(\"RAW_POS.LOCATION\")\n",
        "\n",
        "    t_with_f = truck.join(franchise, truck['FRANCHISE_ID'] == franchise['FRANCHISE_ID'], rsuffix='_f')\n",
        "    oh_w_t_and_l = order_header.join(t_with_f, order_header['TRUCK_ID'] == t_with_f['TRUCK_ID'], rsuffix='_t') \\\n",
        "                                .join(location, order_header['LOCATION_ID'] == location['LOCATION_ID'], rsuffix='_l')\n",
        "    final_df = order_detail.join(oh_w_t_and_l, order_detail['ORDER_ID'] == oh_w_t_and_l['ORDER_ID'], rsuffix='_oh') \\\n",
        "                            .join(menu, order_detail['MENU_ITEM_ID'] == menu['MENU_ITEM_ID'], rsuffix='_m')\n",
        "    final_df = final_df.select(F.col(\"ORDER_ID\"), \\\n",
        "                            F.col(\"TRUCK_ID\"), \\\n",
        "                            F.col(\"ORDER_TS\"), \\\n",
        "                            F.col('ORDER_TS_DATE'), \\\n",
        "                            F.col(\"ORDER_DETAIL_ID\"), \\\n",
        "                            F.col(\"LINE_NUMBER\"), \\\n",
        "                            F.col(\"TRUCK_BRAND_NAME\"), \\\n",
        "                            F.col(\"MENU_TYPE\"), \\\n",
        "                            F.col(\"PRIMARY_CITY\"), \\\n",
        "                            F.col(\"REGION\"), \\\n",
        "                            F.col(\"COUNTRY\"), \\\n",
        "                            F.col(\"FRANCHISE_FLAG\"), \\\n",
        "                            F.col(\"FRANCHISE_ID\"), \\\n",
        "                            F.col(\"FRANCHISEE_FIRST_NAME\"), \\\n",
        "                            F.col(\"FRANCHISEE_LAST_NAME\"), \\\n",
        "                            F.col(\"LOCATION_ID\"), \\\n",
        "                            F.col(\"MENU_ITEM_ID\"), \\\n",
        "                            F.col(\"MENU_ITEM_NAME\"), \\\n",
        "                            F.col(\"QUANTITY\"), \\\n",
        "                            F.col(\"UNIT_PRICE\"), \\\n",
        "                            F.col(\"PRICE\"), \\\n",
        "                            F.col(\"ORDER_AMOUNT\"), \\\n",
        "                            F.col(\"ORDER_TAX_AMOUNT\"), \\\n",
        "                            F.col(\"ORDER_DISCOUNT_AMOUNT\"), \\\n",
        "                            F.col(\"ORDER_TOTAL\"))\n",
        "    final_df.create_or_replace_view('POS_FLATTENED_V')\n",
        "\n",
        "def create_pos_view_stream(session):\n",
        "    session.use_schema('HARMONIZED')\n",
        "    _ = session.sql('CREATE OR REPLACE STREAM POS_FLATTENED_V_STREAM \\\n",
        "                        ON VIEW POS_FLATTENED_V \\\n",
        "                        SHOW_INITIAL_ROWS = TRUE').collect()\n",
        "\n",
        "def test_pos_view(session):\n",
        "    session.use_schema('HARMONIZED')\n",
        "    tv = session.table('POS_FLATTENED_V')\n",
        "    tv.limit(5).show()"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "id": "8f14352d-26ee-40a6-b1f5-0d94ffaa302a",
      "metadata": {
        "codeCollapsed": false,
        "language": "python",
        "name": "cell19"
      },
      "outputs": [],
      "source": [
        "create_pos_view(session)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "id": "78c9076f-0620-4476-a94f-8d2b5897a2d6",
      "metadata": {
        "codeCollapsed": false,
        "language": "python",
        "name": "cell20"
      },
      "outputs": [],
      "source": [
        "create_pos_view_stream(session)"
      ]
    },
    {
      "cell_type": "markdown",
      "id": "419aa2e8-87c8-446f-8236-c8dd2e2440c3",
      "metadata": {
        "name": "cell21"
      },
      "source": [
        "### Snowpark DataFrame API\n",
        "The first thing you'll notice in the `create_pos_view()` function is that we define the Snowflake view using the Snowpark DataFrame API. After defining the final DataFrame, which captures all the logic we want in the view, we can simply call the Snowpark `create_or_replace_view()` method. Here's the final line from the `create_pos_view()` function:\n",
        "\n",
        "```python\n",
        "    final_df.create_or_replace_view('POS_FLATTENED_V')\n",
        "```\n",
        "\n",
        "For more details about the Snowpark Python DataFrame API, please check out our [Working with DataFrames in Snowpark Python](https://docs.snowflake.com/en/developer-guide/snowpark/python/working-with-dataframes.html) page.\n",
        "\n",
        "### Streams for Incremental Processing (CDC)\n",
        "Snowflake makes processing data incrementally very easy. Traditionally the data engineer had to keep track of a high watermark (usually a datetime column) in order to process only new records in a table. This involved tracking and persisting that watermark somewhere and then using it in any query against the source table. But with Snowflake streams all the heavy lifting is done for you by Snowflake. For more details please check out our [Change Tracking Using Table Streams](https://docs.snowflake.com/en/user-guide/streams.html) user guide.\n",
        "\n",
        "All you need to do is create a [`STREAM`](https://docs.snowflake.com/en/sql-reference/sql/create-stream.html) object in Snowflake against your base table or view, then query that stream just like any table in Snowflake. The stream will return only the changed records since the last DML option your performed. To help you work with the changed records, Snowflake streams will supply the following metadata columns along with the base table or view columns:\n",
        "\n",
        "* METADATA$ACTION\n",
        "* METADATA$ISUPDATE\n",
        "* METADATA$ROW_ID\n",
        "\n",
        "For more details about these stream metadata columns please check out the [Stream Columns](https://docs.snowflake.com/en/user-guide/streams-intro.html#stream-columns) section in our documentation.\n",
        "\n",
        "### Streams on views\n",
        "What's really cool about Snowflake's incremental/CDC stream capability is the ability to create a stream on a view! In this example we are creating a stream on a view which joins together 6 of the raw POS tables. Here is the code to do that:\n",
        "\n",
        "```python\n",
        "def create_pos_view_stream(session):\n",
        "    session.use_schema('HARMONIZED')\n",
        "    _ = session.sql('CREATE OR REPLACE STREAM POS_FLATTENED_V_STREAM \\\n",
        "                        ON VIEW POS_FLATTENED_V \\\n",
        "                        SHOW_INITIAL_ROWS = TRUE').collect()\n",
        "```\n",
        "\n",
        "Now when we query the `POS_FLATTENED_V_STREAM` stream to find changed records, Snowflake is actually looking for changed records in any of the 6 tables included in the view. For those who have tried to build incremental/CDC processes around denormalized schemas like this, you will appreciate the incredibly powerful feature that Snowflake provides here.\n",
        "\n",
        "For more details please check out the [Streams on Views](https://docs.snowflake.com/en/user-guide/streams-intro.html#streams-on-views) section in our documentation.\n"
      ]
    },
    {
      "cell_type": "markdown",
      "id": "33f62ff3-5016-4514-8571-ed6195353e07",
      "metadata": {
        "collapsed": false,
        "jupyter": {
          "outputs_hidden": false
        },
        "name": "cell22"
      },
      "source": [
        "## Fahrenheit to Celsius UDF\n",
        "\n",
        "During this step we will be creating and deploying our first Snowpark Python object to Snowflake, a user-defined function (or UDF). To begin with the UDF will be very basic, but in a future step we'll update it to include a third-party Python package. Also in this step you will be introduced to the new SnowCLI, a new developer command line tool. SnowCLI makes building and deploying Snowpark Python objects to Snowflake a consistent experience for the developer. More details below on SnowCLI. To put this in context, we are on step **#5** in our data flow overview."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "id": "a4f247e1-40d0-4003-92d5-3a19712254f1",
      "metadata": {
        "codeCollapsed": false,
        "collapsed": false,
        "jupyter": {
          "outputs_hidden": false
        },
        "language": "sql",
        "name": "cell23"
      },
      "outputs": [],
      "source": [
        "CREATE OR REPLACE FUNCTION NB_HOL_DB.ANALYTICS.FAHRENHEIT_TO_CELSIUS_UDF(\"temp_f\" FLOAT)\n",
        "RETURNS FLOAT\n",
        "LANGUAGE PYTHON\n",
        "RUNTIME_VERSION=3.8\n",
        "HANDLER = 'main'\n",
        "AS '\n",
        "def main(temp_f: float) -> float:\n",
        "    return (float(temp_f) - 32) * (5/9)\n",
        "';"
      ]
    },
    {
      "cell_type": "markdown",
      "id": "09e72e42-4bf6-454c-b76c-e323fcfdbca8",
      "metadata": {
        "name": "cell24"
      },
      "source": [
        "### Running the UDF in Snowflake\n",
        "In order to run the UDF in Snowflake you have a few options. Any UDF in Snowflake can be invoked through SQL as follows:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "id": "070c5e8a-f1e7-4fde-9b8e-e3514e3856e6",
      "metadata": {
        "codeCollapsed": false,
        "collapsed": false,
        "jupyter": {
          "outputs_hidden": false
        },
        "language": "sql",
        "name": "cell25"
      },
      "outputs": [],
      "source": [
        "SELECT NB_HOL_DB.ANALYTICS.FAHRENHEIT_TO_CELSIUS_UDF(35);"
      ]
    },
    {
      "cell_type": "markdown",
      "id": "123bc413-3c14-4092-bc61-d6cc021704e9",
      "metadata": {
        "collapsed": false,
        "jupyter": {
          "outputs_hidden": false
        },
        "name": "cell26"
      },
      "source": [
        "## Orders Update Sproc\n",
        "\n",
        "During this step we will be creating and deploying our first Snowpark Python stored procedure (or sproc) to Snowflake. This sproc will merge changes from the `HARMONIZED.POS_FLATTENED_V_STREAM` stream into our target `HARMONIZED.ORDERS` table. To put this in context, we are on step **#6** in our data flow overview.\n",
        "\n",
        "### Running the Sproc Locally\n",
        "To test the procedure locally, you will execute the following script."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "id": "2e75b295-bd82-4ed6-a565-259a56053c61",
      "metadata": {
        "codeCollapsed": false,
        "collapsed": false,
        "jupyter": {
          "outputs_hidden": false
        },
        "language": "python",
        "name": "cell27"
      },
      "outputs": [],
      "source": [
        "# SNOWFLAKE ADVANTAGE: Python Stored Procedures\n",
        "\n",
        "import time\n",
        "from snowflake.snowpark import Session\n",
        "#import snowflake.snowpark.types as T\n",
        "import snowflake.snowpark.functions as F\n",
        "\n",
        "\n",
        "def table_exists(session, schema='', name=''):\n",
        "    exists = session.sql(\"SELECT EXISTS (SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = '{}' AND TABLE_NAME = '{}') AS TABLE_EXISTS\".format(schema, name)).collect()[0]['TABLE_EXISTS']\n",
        "    return exists\n",
        "\n",
        "def create_orders_table(session):\n",
        "    _ = session.sql(\"CREATE TABLE HARMONIZED.ORDERS LIKE HARMONIZED.POS_FLATTENED_V\").collect()\n",
        "    _ = session.sql(\"ALTER TABLE HARMONIZED.ORDERS ADD COLUMN META_UPDATED_AT TIMESTAMP\").collect()\n",
        "\n",
        "def create_orders_stream(session):\n",
        "    _ = session.sql(\"CREATE STREAM HARMONIZED.ORDERS_STREAM ON TABLE HARMONIZED.ORDERS\").collect()\n",
        "\n",
        "def merge_order_updates(session):\n",
        "    _ = session.sql('ALTER WAREHOUSE HOL_WH SET WAREHOUSE_SIZE = XLARGE WAIT_FOR_COMPLETION = TRUE').collect()\n",
        "\n",
        "    source = session.table('HARMONIZED.POS_FLATTENED_V_STREAM')\n",
        "    target = session.table('HARMONIZED.ORDERS')\n",
        "\n",
        "    # TODO: Is the if clause supposed to be based on \"META_UPDATED_AT\"?\n",
        "    cols_to_update = {c: source[c] for c in source.schema.names if \"METADATA\" not in c}\n",
        "    metadata_col_to_update = {\"META_UPDATED_AT\": F.current_timestamp()}\n",
        "    updates = {**cols_to_update, **metadata_col_to_update}\n",
        "\n",
        "    # merge into DIM_CUSTOMER\n",
        "    target.merge(source, target['ORDER_DETAIL_ID'] == source['ORDER_DETAIL_ID'], \\\n",
        "                        [F.when_matched().update(updates), F.when_not_matched().insert(updates)])\n",
        "\n",
        "    _ = session.sql('ALTER WAREHOUSE HOL_WH SET WAREHOUSE_SIZE = XSMALL').collect()\n",
        "\n",
        "def main(session: Session) -> str:\n",
        "    # Create the ORDERS table and ORDERS_STREAM stream if they don't exist\n",
        "    if not table_exists(session, schema='HARMONIZED', name='ORDERS'):\n",
        "        create_orders_table(session)\n",
        "        create_orders_stream(session)\n",
        "\n",
        "    # Process data incrementally\n",
        "    merge_order_updates(session)\n",
        "#    session.table('HARMONIZED.ORDERS').limit(5).show()\n",
        "\n",
        "    return f\"Successfully processed ORDERS\""
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "id": "f9faffa0-45a3-48e2-aa01-db9c512b0881",
      "metadata": {
        "codeCollapsed": false,
        "collapsed": false,
        "jupyter": {
          "outputs_hidden": false
        },
        "language": "python",
        "name": "cell28"
      },
      "outputs": [],
      "source": [
        "main(session)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "id": "680bb06b-273a-4bba-8155-a4180ec90245",
      "metadata": {
        "codeCollapsed": false,
        "language": "python",
        "name": "cell29"
      },
      "outputs": [],
      "source": [
        "script = '''\n",
        "import time\n",
        "from snowflake.snowpark import Session\n",
        "#import snowflake.snowpark.types as T\n",
        "import snowflake.snowpark.functions as F\n",
        "\n",
        "\n",
        "def table_exists(session, schema='', name=''):\n",
        "    exists = session.sql(\"SELECT EXISTS (SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = '{}' AND TABLE_NAME = '{}') AS TABLE_EXISTS\".format(schema, name)).collect()[0]['TABLE_EXISTS']\n",
        "    return exists\n",
        "\n",
        "def create_orders_table(session):\n",
        "    _ = session.sql(\"CREATE TABLE HARMONIZED.ORDERS LIKE HARMONIZED.POS_FLATTENED_V\").collect()\n",
        "    _ = session.sql(\"ALTER TABLE HARMONIZED.ORDERS ADD COLUMN META_UPDATED_AT TIMESTAMP\").collect()\n",
        "\n",
        "def create_orders_stream(session):\n",
        "    _ = session.sql(\"CREATE STREAM HARMONIZED.ORDERS_STREAM ON TABLE HARMONIZED.ORDERS\").collect()\n",
        "\n",
        "def merge_order_updates(session):\n",
        "    _ = session.sql('ALTER WAREHOUSE HOL_WH SET WAREHOUSE_SIZE = XLARGE WAIT_FOR_COMPLETION = TRUE').collect()\n",
        "\n",
        "    source = session.table('HARMONIZED.POS_FLATTENED_V_STREAM')\n",
        "    target = session.table('HARMONIZED.ORDERS')\n",
        "\n",
        "    # TODO: Is the if clause supposed to be based on \"META_UPDATED_AT\"?\n",
        "    cols_to_update = {c: source[c] for c in source.schema.names if \"METADATA\" not in c}\n",
        "    metadata_col_to_update = {\"META_UPDATED_AT\": F.current_timestamp()}\n",
        "    updates = {**cols_to_update, **metadata_col_to_update}\n",
        "\n",
        "    # merge into DIM_CUSTOMER\n",
        "    target.merge(source, target['ORDER_DETAIL_ID'] == source['ORDER_DETAIL_ID'], \\\n",
        "                        [F.when_matched().update(updates), F.when_not_matched().insert(updates)])\n",
        "\n",
        "    _ = session.sql('ALTER WAREHOUSE HOL_WH SET WAREHOUSE_SIZE = XSMALL').collect()\n",
        "\n",
        "def main(session: Session) -> str:\n",
        "    # Create the ORDERS table and ORDERS_STREAM stream if they don't exist\n",
        "    if not table_exists(session, schema='HARMONIZED', name='ORDERS'):\n",
        "        create_orders_table(session)\n",
        "        create_orders_stream(session)\n",
        "\n",
        "    # Process data incrementally\n",
        "    merge_order_updates(session)\n",
        "#    session.table('HARMONIZED.ORDERS').limit(5).show()\n",
        "\n",
        "    return f\"Successfully processed ORDERS\"\n",
        "'''"
      ]
    },
    {
      "cell_type": "markdown",
      "id": "4de69c48-7dd4-4e15-bb2b-3cd46f8a5972",
      "metadata": {
        "name": "cell30"
      },
      "source": [
        "Here is the SQL query to deploy the procedure:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "id": "597c8215-aa83-4041-b0d0-61918a81b250",
      "metadata": {
        "codeCollapsed": false,
        "language": "sql",
        "name": "cell31"
      },
      "outputs": [],
      "source": [
        "CREATE OR REPLACE PROCEDURE orders_update_sp()\n",
        " RETURNS string\n",
        " LANGUAGE PYTHON\n",
        " RUNTIME_VERSION=3.8\n",
        " PACKAGES=('snowflake-snowpark-python','toml')\n",
        " HANDLER = 'main'\n",
        " AS $$\n",
        " {{script}}\n",
        " $$;"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "id": "edc7aacb-152a-4b53-b83a-a24578b27333",
      "metadata": {
        "codeCollapsed": false,
        "collapsed": false,
        "jupyter": {
          "outputs_hidden": false
        },
        "language": "sql",
        "name": "cell32"
      },
      "outputs": [],
      "source": [
        "CALL ORDERS_UPDATE_SP();"
      ]
    },
    {
      "cell_type": "markdown",
      "id": "46f5f6a9-334c-41be-b69f-72b0b2ff51bd",
      "metadata": {
        "name": "cell33"
      },
      "source": [
        "### More on the Snowpark API\n",
        "In this step we're starting to really use the Snowpark DataFrame API for data transformations. The Snowpark API provides the same functionality as the [Spark SQL API](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/index.html). To begin with you need to create a Snowpark session object. Like PySpark, this is accomplished with the `Session.builder.configs().create()` methods. When running locally, we use the `Session.builder.getOrCreate()` method to create the session object for us. But when deployed to Snowflake, the session object is provisioned for you automatically by Snowflake. And when building a Snowpark Python sproc the contract is that the first argument to the entry point (or handler) function is a Snowpark session.\n",
        "\n",
        "The first thing you'll notice in the script is that we have some functions which use SQL to create objects in Snowflake and to check object status. To issue a SQL statement to Snowflake with the Snowpark API you use the `session.sql()` function, like you'd expect. Here's one example:\n",
        "\n",
        "```python\n",
        "def create_orders_stream(session):\n",
        "    _ = session.sql(\"CREATE STREAM IF NOT EXISTS HARMONIZED.ORDERS_STREAM ON TABLE HARMONIZED.ORDERS \\\n",
        "                    SHOW_INITIAL_ROWS = TRUE;\").collect()\n",
        "```\n",
        "\n",
        "The second thing to point out is how we're using DataFrames to merge changes from the source view to the target table. The Snowpark DataFrame API provides a `merge()` method which will ultimately generate a `MERGE` command in Snowflake.\n",
        "\n",
        "```python\n",
        "    source = session.table('HARMONIZED.POS_FLATTENED_V_STREAM')\n",
        "    target = session.table('HARMONIZED.ORDERS')\n",
        "\n",
        "    # TODO: Is the if clause supposed to be based on \"META_UPDATED_AT\"?\n",
        "    cols_to_update = {c: source[c] for c in source.schema.names if \"METADATA\" not in c}\n",
        "    metadata_col_to_update = {\"META_UPDATED_AT\": F.current_timestamp()}\n",
        "    updates = {**cols_to_update, **metadata_col_to_update}\n",
        "\n",
        "    # merge into DIM_CUSTOMER\n",
        "    target.merge(source, target['ORDER_DETAIL_ID'] == source['ORDER_DETAIL_ID'], \\\n",
        "                        [F.when_matched().update(updates), F.when_not_matched().insert(updates)])\n",
        "```\n",
        "\n",
        "Again, for more details about the Snowpark Python DataFrame API, please check out our [Working with DataFrames in Snowpark Python](https://docs.snowflake.com/en/developer-guide/snowpark/python/working-with-dataframes.html) page.\n"
      ]
    },
    {
      "cell_type": "markdown",
      "id": "fd90d2f2-a045-4425-8ad0-7533c7d7611c",
      "metadata": {
        "collapsed": false,
        "jupyter": {
          "outputs_hidden": false
        },
        "name": "cell34"
      },
      "source": [
        "## Daily City Metrics Update Sproc\n",
        "During this step we will be creating and deploying our second Snowpark Python sproc to Snowflake. This sproc will join the `HARMONIZED.ORDERS` data with the Weather Source data to create a final, aggregated table for analysis named `ANALYTICS.DAILY_CITY_METRICS`. We will process the data incrementally from the `HARMONIZED.ORDERS` table using another Snowflake Stream. And we will again use the Snowpark DataFrame `merge()` method to merge/upsert the data. To put this in context, we are on step **#7** in our data flow overview."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "id": "954fa22c-f646-4ab8-aea4-6ac17e5b96e3",
      "metadata": {
        "codeCollapsed": false,
        "collapsed": false,
        "jupyter": {
          "outputs_hidden": false
        },
        "language": "python",
        "name": "cell35"
      },
      "outputs": [],
      "source": [
        "import time\n",
        "from snowflake.snowpark import Session\n",
        "import snowflake.snowpark.types as T\n",
        "import snowflake.snowpark.functions as F\n",
        "\n",
        "\n",
        "def table_exists(session, schema='', name=''):\n",
        "    exists = session.sql(\"SELECT EXISTS (SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = '{}' AND TABLE_NAME = '{}') AS TABLE_EXISTS\".format(schema, name)).collect()[0]['TABLE_EXISTS']\n",
        "    return exists\n",
        "\n",
        "def create_daily_city_metrics_table(session):\n",
        "    SHARED_COLUMNS= [T.StructField(\"DATE\", T.DateType()),\n",
        "                                        T.StructField(\"CITY_NAME\", T.StringType()),\n",
        "                                        T.StructField(\"COUNTRY_DESC\", T.StringType()),\n",
        "                                        T.StructField(\"DAILY_SALES\", T.StringType()),\n",
        "                                        T.StructField(\"AVG_TEMPERATURE_FAHRENHEIT\", T.DecimalType()),\n",
        "                                        T.StructField(\"AVG_TEMPERATURE_CELSIUS\", T.DecimalType()),\n",
        "                                        T.StructField(\"AVG_PRECIPITATION_INCHES\", T.DecimalType()),\n",
        "                                        T.StructField(\"AVG_PRECIPITATION_MILLIMETERS\", T.DecimalType()),\n",
        "                                        T.StructField(\"MAX_WIND_SPEED_100M_MPH\", T.DecimalType()),\n",
        "                                    ]\n",
        "    DAILY_CITY_METRICS_COLUMNS = [*SHARED_COLUMNS, T.StructField(\"META_UPDATED_AT\", T.TimestampType())]\n",
        "    DAILY_CITY_METRICS_SCHEMA = T.StructType(DAILY_CITY_METRICS_COLUMNS)\n",
        "\n",
        "    dcm = session.create_dataframe([[None]*len(DAILY_CITY_METRICS_SCHEMA.names)], schema=DAILY_CITY_METRICS_SCHEMA) \\\n",
        "                        .na.drop() \\\n",
        "                        .write.mode('overwrite').save_as_table('ANALYTICS.DAILY_CITY_METRICS')\n",
        "    dcm = session.table('ANALYTICS.DAILY_CITY_METRICS')\n",
        "\n",
        "\n",
        "def merge_daily_city_metrics(session):\n",
        "    _ = session.sql('ALTER WAREHOUSE HOL_WH SET WAREHOUSE_SIZE = XLARGE WAIT_FOR_COMPLETION = TRUE').collect()\n",
        "\n",
        "    print(\"{} records in stream\".format(session.table('HARMONIZED.ORDERS_STREAM').count()))\n",
        "    orders_stream_dates = session.table('HARMONIZED.ORDERS_STREAM').select(F.col(\"ORDER_TS_DATE\").alias(\"DATE\")).distinct()\n",
        "    orders_stream_dates.limit(5).show()\n",
        "\n",
        "    orders = session.table(\"HARMONIZED.ORDERS_STREAM\").group_by(F.col('ORDER_TS_DATE'), F.col('PRIMARY_CITY'), F.col('COUNTRY')) \\\n",
        "                                        .agg(F.sum(F.col(\"PRICE\")).as_(\"price_nulls\")) \\\n",
        "                                        .with_column(\"DAILY_SALES\", F.call_builtin(\"ZEROIFNULL\", F.col(\"price_nulls\"))) \\\n",
        "                                        .select(F.col('ORDER_TS_DATE').alias(\"DATE\"), F.col(\"PRIMARY_CITY\").alias(\"CITY_NAME\"), \\\n",
        "                                        F.col(\"COUNTRY\").alias(\"COUNTRY_DESC\"), F.col(\"DAILY_SALES\"))\n",
        "#    orders.limit(5).show()\n",
        "\n",
        "    weather_pc = session.table(\"FROSTBYTE_WEATHERSOURCE.ONPOINT_ID.POSTAL_CODES\")\n",
        "    countries = session.table(\"RAW_POS.COUNTRY\")\n",
        "    weather = session.table(\"FROSTBYTE_WEATHERSOURCE.ONPOINT_ID.HISTORY_DAY\")\n",
        "    weather = weather.join(weather_pc, (weather['POSTAL_CODE'] == weather_pc['POSTAL_CODE']) & (weather['COUNTRY'] == weather_pc['COUNTRY']), rsuffix='_pc')\n",
        "    weather = weather.join(countries, (weather['COUNTRY'] == countries['ISO_COUNTRY']) & (weather['CITY_NAME'] == countries['CITY']), rsuffix='_c')\n",
        "    weather = weather.join(orders_stream_dates, weather['DATE_VALID_STD'] == orders_stream_dates['DATE'])\n",
        "\n",
        "    weather_agg = weather.group_by(F.col('DATE_VALID_STD'), F.col('CITY_NAME'), F.col('COUNTRY_C')) \\\n",
        "                        .agg( \\\n",
        "                            F.avg('AVG_TEMPERATURE_AIR_2M_F').alias(\"AVG_TEMPERATURE_F\"), \\\n",
        "                            F.avg(F.call_udf(\"ANALYTICS.FAHRENHEIT_TO_CELSIUS_UDF\", F.col(\"AVG_TEMPERATURE_AIR_2M_F\"))).alias(\"AVG_TEMPERATURE_C\"), \\\n",
        "                            F.avg(\"TOT_PRECIPITATION_IN\").alias(\"AVG_PRECIPITATION_IN\"), \\\n",
        "                            F.avg(F.call_udf(\"ANALYTICS.INCH_TO_MILLIMETER_UDF\", F.col(\"TOT_PRECIPITATION_IN\"))).alias(\"AVG_PRECIPITATION_MM\"), \\\n",
        "                            F.max(F.col(\"MAX_WIND_SPEED_100M_MPH\")).alias(\"MAX_WIND_SPEED_100M_MPH\") \\\n",
        "                        ) \\\n",
        "                        .select(F.col(\"DATE_VALID_STD\").alias(\"DATE\"), F.col(\"CITY_NAME\"), F.col(\"COUNTRY_C\").alias(\"COUNTRY_DESC\"), \\\n",
        "                            F.round(F.col(\"AVG_TEMPERATURE_F\"), 2).alias(\"AVG_TEMPERATURE_FAHRENHEIT\"), \\\n",
        "                            F.round(F.col(\"AVG_TEMPERATURE_C\"), 2).alias(\"AVG_TEMPERATURE_CELSIUS\"), \\\n",
        "                            F.round(F.col(\"AVG_PRECIPITATION_IN\"), 2).alias(\"AVG_PRECIPITATION_INCHES\"), \\\n",
        "                            F.round(F.col(\"AVG_PRECIPITATION_MM\"), 2).alias(\"AVG_PRECIPITATION_MILLIMETERS\"), \\\n",
        "                            F.col(\"MAX_WIND_SPEED_100M_MPH\")\n",
        "                            )\n",
        "#    weather_agg.limit(5).show()\n",
        "\n",
        "    daily_city_metrics_stg = orders.join(weather_agg, (orders['DATE'] == weather_agg['DATE']) & (orders['CITY_NAME'] == weather_agg['CITY_NAME']) & (orders['COUNTRY_DESC'] == weather_agg['COUNTRY_DESC']), \\\n",
        "                        how='left', rsuffix='_w') \\\n",
        "                    .select(\"DATE\", \"CITY_NAME\", \"COUNTRY_DESC\", \"DAILY_SALES\", \\\n",
        "                        \"AVG_TEMPERATURE_FAHRENHEIT\", \"AVG_TEMPERATURE_CELSIUS\", \\\n",
        "                        \"AVG_PRECIPITATION_INCHES\", \"AVG_PRECIPITATION_MILLIMETERS\", \\\n",
        "                        \"MAX_WIND_SPEED_100M_MPH\")\n",
        "#    daily_city_metrics_stg.limit(5).show()\n",
        "\n",
        "    cols_to_update = {c: daily_city_metrics_stg[c] for c in daily_city_metrics_stg.schema.names}\n",
        "    metadata_col_to_update = {\"META_UPDATED_AT\": F.current_timestamp()}\n",
        "    updates = {**cols_to_update, **metadata_col_to_update}\n",
        "\n",
        "    dcm = session.table('ANALYTICS.DAILY_CITY_METRICS')\n",
        "    dcm.merge(daily_city_metrics_stg, (dcm['DATE'] == daily_city_metrics_stg['DATE']) & (dcm['CITY_NAME'] == daily_city_metrics_stg['CITY_NAME']) & (dcm['COUNTRY_DESC'] == daily_city_metrics_stg['COUNTRY_DESC']), \\\n",
        "                        [F.when_matched().update(updates), F.when_not_matched().insert(updates)])\n",
        "\n",
        "    _ = session.sql('ALTER WAREHOUSE HOL_WH SET WAREHOUSE_SIZE = XSMALL').collect()\n",
        "\n",
        "def main(session: Session) -> str:\n",
        "    # Create the DAILY_CITY_METRICS table if it doesn't exist\n",
        "    if not table_exists(session, schema='ANALYTICS', name='DAILY_CITY_METRICS'):\n",
        "        create_daily_city_metrics_table(session)\n",
        "    \n",
        "    merge_daily_city_metrics(session)\n",
        "#    session.table('ANALYTICS.DAILY_CITY_METRICS').limit(5).show()\n",
        "\n",
        "    return f\"Successfully processed DAILY_CITY_METRICS\""
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "id": "1ddce8d6-8d33-4ec5-ba24-6ce352f3ee0e",
      "metadata": {
        "codeCollapsed": false,
        "collapsed": false,
        "jupyter": {
          "outputs_hidden": false
        },
        "language": "python",
        "name": "cell36"
      },
      "outputs": [],
      "source": [
        "main(session)"
      ]
    },
    {
      "cell_type": "markdown",
      "id": "cfadb325-1f5a-4ce5-baef-8643bb22dd03",
      "metadata": {
        "collapsed": false,
        "name": "cell37"
      },
      "source": [
        "### Deploying the Sproc to Snowflake\n",
        "To deploy your sproc to Snowflake by running the following commands:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "id": "04c908df-3f1c-49ae-97e5-b89e159bd44b",
      "metadata": {
        "codeCollapsed": false,
        "language": "python",
        "name": "cell38"
      },
      "outputs": [],
      "source": [
        "script = '''\n",
        "import time\n",
        "from snowflake.snowpark import Session\n",
        "import snowflake.snowpark.types as T\n",
        "import snowflake.snowpark.functions as F\n",
        "\n",
        "\n",
        "def table_exists(session, schema='', name=''):\n",
        "    exists = session.sql(\"SELECT EXISTS (SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = '{}' AND TABLE_NAME = '{}') AS TABLE_EXISTS\".format(schema, name)).collect()[0]['TABLE_EXISTS']\n",
        "    return exists\n",
        "\n",
        "def create_daily_city_metrics_table(session):\n",
        "    SHARED_COLUMNS= [T.StructField(\"DATE\", T.DateType()),\n",
        "                                        T.StructField(\"CITY_NAME\", T.StringType()),\n",
        "                                        T.StructField(\"COUNTRY_DESC\", T.StringType()),\n",
        "                                        T.StructField(\"DAILY_SALES\", T.StringType()),\n",
        "                                        T.StructField(\"AVG_TEMPERATURE_FAHRENHEIT\", T.DecimalType()),\n",
        "                                        T.StructField(\"AVG_TEMPERATURE_CELSIUS\", T.DecimalType()),\n",
        "                                        T.StructField(\"AVG_PRECIPITATION_INCHES\", T.DecimalType()),\n",
        "                                        T.StructField(\"AVG_PRECIPITATION_MILLIMETERS\", T.DecimalType()),\n",
        "                                        T.StructField(\"MAX_WIND_SPEED_100M_MPH\", T.DecimalType()),\n",
        "                                    ]\n",
        "    DAILY_CITY_METRICS_COLUMNS = [*SHARED_COLUMNS, T.StructField(\"META_UPDATED_AT\", T.TimestampType())]\n",
        "    DAILY_CITY_METRICS_SCHEMA = T.StructType(DAILY_CITY_METRICS_COLUMNS)\n",
        "\n",
        "    dcm = session.create_dataframe([[None]*len(DAILY_CITY_METRICS_SCHEMA.names)], schema=DAILY_CITY_METRICS_SCHEMA) \\\n",
        "                        .na.drop() \\\n",
        "                        .write.mode('overwrite').save_as_table('ANALYTICS.DAILY_CITY_METRICS')\n",
        "    dcm = session.table('ANALYTICS.DAILY_CITY_METRICS')\n",
        "\n",
        "\n",
        "def merge_daily_city_metrics(session):\n",
        "    _ = session.sql('ALTER WAREHOUSE HOL_WH SET WAREHOUSE_SIZE = XLARGE WAIT_FOR_COMPLETION = TRUE').collect()\n",
        "\n",
        "    print(\"{} records in stream\".format(session.table('HARMONIZED.ORDERS_STREAM').count()))\n",
        "    orders_stream_dates = session.table('HARMONIZED.ORDERS_STREAM').select(F.col(\"ORDER_TS_DATE\").alias(\"DATE\")).distinct()\n",
        "    orders_stream_dates.limit(5).show()\n",
        "\n",
        "    orders = session.table(\"HARMONIZED.ORDERS_STREAM\").group_by(F.col('ORDER_TS_DATE'), F.col('PRIMARY_CITY'), F.col('COUNTRY')) \\\n",
        "                                        .agg(F.sum(F.col(\"PRICE\")).as_(\"price_nulls\")) \\\n",
        "                                        .with_column(\"DAILY_SALES\", F.call_builtin(\"ZEROIFNULL\", F.col(\"price_nulls\"))) \\\n",
        "                                        .select(F.col('ORDER_TS_DATE').alias(\"DATE\"), F.col(\"PRIMARY_CITY\").alias(\"CITY_NAME\"), \\\n",
        "                                        F.col(\"COUNTRY\").alias(\"COUNTRY_DESC\"), F.col(\"DAILY_SALES\"))\n",
        "#    orders.limit(5).show()\n",
        "\n",
        "    weather_pc = session.table(\"FROSTBYTE_WEATHERSOURCE.ONPOINT_ID.POSTAL_CODES\")\n",
        "    countries = session.table(\"RAW_POS.COUNTRY\")\n",
        "    weather = session.table(\"FROSTBYTE_WEATHERSOURCE.ONPOINT_ID.HISTORY_DAY\")\n",
        "    weather = weather.join(weather_pc, (weather['POSTAL_CODE'] == weather_pc['POSTAL_CODE']) & (weather['COUNTRY'] == weather_pc['COUNTRY']), rsuffix='_pc')\n",
        "    weather = weather.join(countries, (weather['COUNTRY'] == countries['ISO_COUNTRY']) & (weather['CITY_NAME'] == countries['CITY']), rsuffix='_c')\n",
        "    weather = weather.join(orders_stream_dates, weather['DATE_VALID_STD'] == orders_stream_dates['DATE'])\n",
        "\n",
        "    weather_agg = weather.group_by(F.col('DATE_VALID_STD'), F.col('CITY_NAME'), F.col('COUNTRY_C')) \\\n",
        "                        .agg( \\\n",
        "                            F.avg('AVG_TEMPERATURE_AIR_2M_F').alias(\"AVG_TEMPERATURE_F\"), \\\n",
        "                            F.avg(F.call_udf(\"ANALYTICS.FAHRENHEIT_TO_CELSIUS_UDF\", F.col(\"AVG_TEMPERATURE_AIR_2M_F\"))).alias(\"AVG_TEMPERATURE_C\"), \\\n",
        "                            F.avg(\"TOT_PRECIPITATION_IN\").alias(\"AVG_PRECIPITATION_IN\"), \\\n",
        "                            F.avg(F.call_udf(\"ANALYTICS.INCH_TO_MILLIMETER_UDF\", F.col(\"TOT_PRECIPITATION_IN\"))).alias(\"AVG_PRECIPITATION_MM\"), \\\n",
        "                            F.max(F.col(\"MAX_WIND_SPEED_100M_MPH\")).alias(\"MAX_WIND_SPEED_100M_MPH\") \\\n",
        "                        ) \\\n",
        "                        .select(F.col(\"DATE_VALID_STD\").alias(\"DATE\"), F.col(\"CITY_NAME\"), F.col(\"COUNTRY_C\").alias(\"COUNTRY_DESC\"), \\\n",
        "                            F.round(F.col(\"AVG_TEMPERATURE_F\"), 2).alias(\"AVG_TEMPERATURE_FAHRENHEIT\"), \\\n",
        "                            F.round(F.col(\"AVG_TEMPERATURE_C\"), 2).alias(\"AVG_TEMPERATURE_CELSIUS\"), \\\n",
        "                            F.round(F.col(\"AVG_PRECIPITATION_IN\"), 2).alias(\"AVG_PRECIPITATION_INCHES\"), \\\n",
        "                            F.round(F.col(\"AVG_PRECIPITATION_MM\"), 2).alias(\"AVG_PRECIPITATION_MILLIMETERS\"), \\\n",
        "                            F.col(\"MAX_WIND_SPEED_100M_MPH\")\n",
        "                            )\n",
        "#    weather_agg.limit(5).show()\n",
        "\n",
        "    daily_city_metrics_stg = orders.join(weather_agg, (orders['DATE'] == weather_agg['DATE']) & (orders['CITY_NAME'] == weather_agg['CITY_NAME']) & (orders['COUNTRY_DESC'] == weather_agg['COUNTRY_DESC']), \\\n",
        "                        how='left', rsuffix='_w') \\\n",
        "                    .select(\"DATE\", \"CITY_NAME\", \"COUNTRY_DESC\", \"DAILY_SALES\", \\\n",
        "                        \"AVG_TEMPERATURE_FAHRENHEIT\", \"AVG_TEMPERATURE_CELSIUS\", \\\n",
        "                        \"AVG_PRECIPITATION_INCHES\", \"AVG_PRECIPITATION_MILLIMETERS\", \\\n",
        "                        \"MAX_WIND_SPEED_100M_MPH\")\n",
        "#    daily_city_metrics_stg.limit(5).show()\n",
        "\n",
        "    cols_to_update = {c: daily_city_metrics_stg[c] for c in daily_city_metrics_stg.schema.names}\n",
        "    metadata_col_to_update = {\"META_UPDATED_AT\": F.current_timestamp()}\n",
        "    updates = {**cols_to_update, **metadata_col_to_update}\n",
        "\n",
        "    dcm = session.table('ANALYTICS.DAILY_CITY_METRICS')\n",
        "    dcm.merge(daily_city_metrics_stg, (dcm['DATE'] == daily_city_metrics_stg['DATE']) & (dcm['CITY_NAME'] == daily_city_metrics_stg['CITY_NAME']) & (dcm['COUNTRY_DESC'] == daily_city_metrics_stg['COUNTRY_DESC']), \\\n",
        "                        [F.when_matched().update(updates), F.when_not_matched().insert(updates)])\n",
        "\n",
        "    _ = session.sql('ALTER WAREHOUSE HOL_WH SET WAREHOUSE_SIZE = XSMALL').collect()\n",
        "\n",
        "def main(session: Session) -> str:\n",
        "    # Create the DAILY_CITY_METRICS table if it doesn't exist\n",
        "    if not table_exists(session, schema='ANALYTICS', name='DAILY_CITY_METRICS'):\n",
        "        create_daily_city_metrics_table(session)\n",
        "    \n",
        "    merge_daily_city_metrics(session)\n",
        "#    session.table('ANALYTICS.DAILY_CITY_METRICS').limit(5).show()\n",
        "\n",
        "    return f\"Successfully processed DAILY_CITY_METRICS\"\n",
        "'''"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "id": "02b166f6-97d8-4d08-8c41-f49c0aae1aec",
      "metadata": {
        "codeCollapsed": false,
        "language": "sql",
        "name": "cell39"
      },
      "outputs": [],
      "source": [
        "CREATE OR REPLACE PROCEDURE DAILY_CITY_METRICS_UPDATE_SP()\n",
        " RETURNS string\n",
        " LANGUAGE PYTHON\n",
        " RUNTIME_VERSION=3.8\n",
        " PACKAGES=('snowflake-snowpark-python','toml')\n",
        " HANDLER = 'main'\n",
        " AS $$\n",
        " {{script}}\n",
        " $$;"
      ]
    },
    {
      "cell_type": "markdown",
      "id": "b9446717-c53a-41f2-9a80-6ebc5548ad9c",
      "metadata": {
        "name": "cell40"
      },
      "source": [
        "### Running the Sproc in Snowflake\n",
        "In order to run the sproc in Snowflake you have a few options. Any sproc in Snowflake can be invoked through SQL as follows:\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "id": "16e5289e-5197-4bda-ac42-551c609787b5",
      "metadata": {
        "codeCollapsed": false,
        "collapsed": false,
        "jupyter": {
          "outputs_hidden": false
        },
        "language": "sql",
        "name": "cell41"
      },
      "outputs": [],
      "source": [
        "CALL DAILY_CITY_METRICS_UPDATE_SP();"
      ]
    },
    {
      "cell_type": "markdown",
      "id": "59c9f51b-0ba8-40ee-9524-a52086cafc56",
      "metadata": {
        "name": "cell42"
      },
      "source": [
        "### Data Modeling Best Practice\n",
        "When modeling data for analysis a best practice has been to clearly define and manage the schema of the table. In step 2, when we loaded raw data from Parquet we took advantage of Snowflake's schema detection feature to create a table with the same schema as the Parquet files. In this step we are explicitly defining the schema in DataFrame syntax and using that to create the table.\n",
        "\n",
        "```python\n",
        "def create_daily_city_metrics_table(session):\n",
        "    SHARED_COLUMNS= [T.StructField(\"DATE\", T.DateType()),\n",
        "                                        T.StructField(\"CITY_NAME\", T.StringType()),\n",
        "                                        T.StructField(\"COUNTRY_DESC\", T.StringType()),\n",
        "                                        T.StructField(\"DAILY_SALES\", T.StringType()),\n",
        "                                        T.StructField(\"AVG_TEMPERATURE_FAHRENHEIT\", T.DecimalType()),\n",
        "                                        T.StructField(\"AVG_TEMPERATURE_CELSIUS\", T.DecimalType()),\n",
        "                                        T.StructField(\"AVG_PRECIPITATION_INCHES\", T.DecimalType()),\n",
        "                                        T.StructField(\"AVG_PRECIPITATION_MILLIMETERS\", T.DecimalType()),\n",
        "                                        T.StructField(\"MAX_WIND_SPEED_100M_MPH\", T.DecimalType()),\n",
        "                                    ]\n",
        "    DAILY_CITY_METRICS_COLUMNS = [*SHARED_COLUMNS, T.StructField(\"META_UPDATED_AT\", T.TimestampType())]\n",
        "    DAILY_CITY_METRICS_SCHEMA = T.StructType(DAILY_CITY_METRICS_COLUMNS)\n",
        "\n",
        "    dcm = session.create_dataframe([[None]*len(DAILY_CITY_METRICS_SCHEMA.names)], schema=DAILY_CITY_METRICS_SCHEMA) \\\n",
        "                        .na.drop() \\\n",
        "                        .write.mode('overwrite').save_as_table('ANALYTICS.DAILY_CITY_METRICS')\n",
        "    dcm = session.table('ANALYTICS.DAILY_CITY_METRICS')\n",
        "```\n",
        "\n",
        "### Complex Aggregation Query\n",
        "The `merge_daily_city_metrics()` function contains a complex aggregation query which is used to join together and aggregate the data from our POS and Weather Source. Take a look at the series of complex series of joins and aggregations that are expressed, and how we're even leveraging the Snowpark UDF we created in step #5!\n",
        "\n",
        "The complex aggregation query is then merged into the final analytics table using the Snowpark `merge()` method. If you haven't already, check out your Snowflake Query history and see which queries were generated by the Snowpark API. In this case you will see that the Snowpark API took all the complex logic, including the merge and created a single Snowflake query to execute!\n"
      ]
    },
    {
      "cell_type": "markdown",
      "id": "3e8c2c63-c904-46aa-b0e4-dddaa502d5a4",
      "metadata": {
        "collapsed": false,
        "jupyter": {
          "outputs_hidden": false
        },
        "name": "cell43"
      },
      "source": [
        "# Orchestrate Jobs\n",
        "\n",
        "During this step we will be orchestrating our new Snowpark pipelines with Snowflake's native orchestration feature named Tasks. We will create two tasks, one for each stored procedure, and chain them together. We will then run the tasks. To put this in context, we are on step **#8** in our data flow overview.\n"
      ]
    },
    {
      "cell_type": "markdown",
      "id": "d5b5b07a-cf8b-414b-9308-e1668ed57746",
      "metadata": {
        "name": "cell44"
      },
      "source": [
        "### Run the Script"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "id": "868473c7-aea4-430a-be35-7f710fc34729",
      "metadata": {
        "codeCollapsed": false,
        "collapsed": false,
        "jupyter": {
          "outputs_hidden": false
        },
        "language": "sql",
        "name": "cell45"
      },
      "outputs": [],
      "source": [
        "-- ----------------------------------------------------------------------------\n",
        "-- Step #1: Create the tasks to call our Python stored procedures\n",
        "-- ----------------------------------------------------------------------------\n",
        "\n",
        "CREATE OR REPLACE TASK ORDERS_UPDATE_TASK\n",
        "WAREHOUSE = HOL_WH\n",
        "WHEN\n",
        "  SYSTEM$STREAM_HAS_DATA('POS_FLATTENED_V_STREAM')\n",
        "AS\n",
        "CALL HARMONIZED.ORDERS_UPDATE_SP();\n",
        "\n",
        "CREATE OR REPLACE TASK DAILY_CITY_METRICS_UPDATE_TASK\n",
        "WAREHOUSE = HOL_WH\n",
        "AFTER ORDERS_UPDATE_TASK\n",
        "WHEN\n",
        "  SYSTEM$STREAM_HAS_DATA('ORDERS_STREAM')\n",
        "AS\n",
        "CALL ANALYTICS.DAILY_CITY_METRICS_UPDATE_SP();\n",
        "\n",
        "-- ----------------------------------------------------------------------------\n",
        "-- Step #2: Execute the tasks\n",
        "-- ----------------------------------------------------------------------------\n",
        "\n",
        "ALTER TASK DAILY_CITY_METRICS_UPDATE_TASK RESUME;\n",
        "\n",
        "EXECUTE TASK ORDERS_UPDATE_TASK;"
      ]
    },
    {
      "cell_type": "markdown",
      "id": "249ba473-5fb5-4ecd-a99c-99c1dcc10f0e",
      "metadata": {
        "collapsed": false,
        "name": "cell46"
      },
      "source": [
        "### Running the Tasks\n",
        "In this step we did not create a schedule for our task DAG, so it will not run on its own at this point. So in this script you will notice that we manually execute the DAG, like this:\n",
        "\n",
        "```sql\n",
        "EXECUTE TASK ORDERS_UPDATE_TASK;\n",
        "```\n",
        "\n",
        "To see what happened when you ran this task just now, highlight and run (using CMD/CTRL+Enter) this commented query in the script:\n",
        "\n",
        "```sql\n",
        "SELECT *\n",
        "FROM TABLE(INFORMATION_SCHEMA.TASK_HISTORY(\n",
        "    SCHEDULED_TIME_RANGE_START=>DATEADD('DAY',-1,CURRENT_TIMESTAMP()),\n",
        "    RESULT_LIMIT => 100))\n",
        "ORDER BY SCHEDULED_TIME DESC\n",
        ";\n",
        "```\n",
        "\n",
        "You will notice in the task history output that it skipped our task `ORDERS_UPDATE_TASK`. This is correct, because our `HARMONIZED.POS_FLATTENED_V_STREAM` stream doesn't have any data. We'll add some new data and run them again in the next step.\n",
        "\n",
        "### More on Tasks\n",
        "Tasks are Snowflake's native scheduling/orchestration feature. With a task you can execute any one of the following types of SQL code:\n",
        "\n",
        "* Single SQL statement\n",
        "* Call to a stored procedure\n",
        "* Procedural logic using Snowflake Scripting Developer Guide\n",
        "\n",
        "For this Quickstart we'll call our Snowpark stored procedures. Here is the SQL DDL code to create the second task:\n",
        "\n",
        "```sql\n",
        "CREATE OR REPLACE TASK DAILY_CITY_METRICS_UPDATE_TASK\n",
        "WAREHOUSE = HOL_WH\n",
        "AFTER ORDERS_UPDATE_TASK\n",
        "WHEN\n",
        "  SYSTEM$STREAM_HAS_DATA('ORDERS_STREAM')\n",
        "AS\n",
        "CALL ANALYTICS.DAILY_CITY_METRICS_UPDATE_SP();\n",
        "```\n",
        "\n",
        "A few things to point out. First you specify which Snowflake virtual warehouse to use when running the task with the `WAREHOUSE` clause. The `AFTER` clause lets you define the relationship between tasks, and the structure of this relationship is a Directed Acyclic Graph (or DAG) like most orchestration tools provide. The `AS` clause let's you define what the task should do when it runs, in this case to call our stored procedure.\n",
        "\n",
        "The `WHEN` clause is really cool. We've already seen how streams work in Snowflake by allowing you to incrementally process data. We've even seen how you can create a stream on a view (which joins many tables together) and create a stream on that view to process its data incrementally! Here in the `WHEN` clause we're calling a system function `SYSTEM$STREAM_HAS_DATA()` which returns true if the specified stream has new data. With the `WHEN` clause in place the virtual warehouse will only be started up when the stream has new data. So if there's no new data when the task runs then your warehouse won't be started up and you won't be charged. You will only be charged when there's new data to process. Pretty cool, huh?\n",
        "\n",
        "As mentioned above we did not define a `SCHEDULE` for the root task, so this DAG will not run on its own. That's fine for this Quickstart, but in a real situation you would define a schedule. See [CREATE TASK](https://docs.snowflake.com/en/sql-reference/sql/create-task.html) for the details.\n",
        "\n",
        "And for more details on Tasks see [Introduction to Tasks](https://docs.snowflake.com/en/user-guide/tasks-intro.html).\n",
        "\n",
        "### Task Metadata\n",
        "Snowflake keeps metadata for almost everything you do, and makes that metadata available for you to query (and to create any type of process around). Tasks are no different, Snowflake maintains rich metadata to help you monitor your task runs. Here are a few sample SQL queries you can use to monitor your tasks runs:\n",
        "\n",
        "```sql\n",
        "-- Get a list of tasks\n",
        "SHOW TASKS;\n",
        "\n",
        "-- Task execution history in the past day\n",
        "SELECT *\n",
        "FROM TABLE(INFORMATION_SCHEMA.TASK_HISTORY(\n",
        "    SCHEDULED_TIME_RANGE_START=>DATEADD('DAY',-1,CURRENT_TIMESTAMP()),\n",
        "    RESULT_LIMIT => 100))\n",
        "ORDER BY SCHEDULED_TIME DESC\n",
        ";\n",
        "\n",
        "-- Scheduled task runs\n",
        "SELECT\n",
        "    TIMESTAMPDIFF(SECOND, CURRENT_TIMESTAMP, SCHEDULED_TIME) NEXT_RUN,\n",
        "    SCHEDULED_TIME,\n",
        "    NAME,\n",
        "    STATE\n",
        "FROM TABLE(INFORMATION_SCHEMA.TASK_HISTORY())\n",
        "WHERE STATE = 'SCHEDULED'\n",
        "ORDER BY COMPLETED_TIME DESC;\n",
        "```\n",
        "\n",
        "### Monitoring Tasks\n",
        "So while you're free to create any operational or monitoring process you wish, Snowflake provides some rich task observability features in our Snowsight UI. Try it out for yourself by following these steps:\n",
        "\n",
        "1. In the Snowsight navigation menu, click **Data** \u00bb **Databases**.\n",
        "1. In the right pane, using the object explorer, navigate to a database and schema.\n",
        "1. For the selected schema, select and expand **Tasks**.\n",
        "1. Select a task. Task information is displayed, including **Task Details**, **Graph**, and **Run History** sub-tabs.\n",
        "1. Select the **Graph** tab. The task graph appears, displaying a hierarchy of child tasks.\n",
        "1. Select a task to view its details."
      ]
    },
    {
      "cell_type": "markdown",
      "id": "0439490a-61bd-45cd-b99b-357de4c86efa",
      "metadata": {
        "collapsed": false,
        "jupyter": {
          "outputs_hidden": false
        },
        "name": "cell47"
      },
      "source": [
        "# Process Incrementally\n",
        "\n",
        "During this step we will be adding new data to our POS order tables and then running our entire end-to-end pipeline to process the new data. And this entire pipeline will be processing data incrementally thanks to Snowflake's advanced stream/CDC capabilities. To put this in context, we are on step **#9** in our data flow overview.\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "id": "c8084fdd-ad00-4d02-b762-bbd076d17bd3",
      "metadata": {
        "codeCollapsed": false,
        "collapsed": false,
        "language": "sql",
        "name": "cell48"
      },
      "outputs": [],
      "source": [
        "-- ----------------------------------------------------------------------------\n",
        "-- Step #1: Add new/remaining order data\n",
        "-- ----------------------------------------------------------------------------\n",
        "\n",
        "USE SCHEMA RAW_POS;\n",
        "\n",
        "ALTER WAREHOUSE HOL_WH SET WAREHOUSE_SIZE = XLARGE WAIT_FOR_COMPLETION = TRUE;\n",
        "\n",
        "LS @external.frostbyte_raw_stage/pos/order_header/year=2022"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "id": "e7438f98-90f1-4a50-bbd8-832ce084a440",
      "metadata": {
        "codeCollapsed": false,
        "collapsed": false,
        "language": "sql",
        "name": "cell49"
      },
      "outputs": [],
      "source": [
        "COPY INTO ORDER_HEADER FROM @external.frostbyte_raw_stage/pos/order_header/year=2022\n",
        "FILE_FORMAT = (FORMAT_NAME = EXTERNAL.PARQUET_FORMAT)\n",
        "MATCH_BY_COLUMN_NAME = CASE_SENSITIVE;"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "id": "62ac79c7-a351-41e2-b80a-96c2d899c363",
      "metadata": {
        "language": "sql",
        "name": "cell50"
      },
      "outputs": [],
      "source": [
        "COPY INTO ORDER_DETAIL FROM @external.frostbyte_raw_stage/pos/order_detail/year=2022\n",
        "FILE_FORMAT = (FORMAT_NAME = EXTERNAL.PARQUET_FORMAT)\n",
        "MATCH_BY_COLUMN_NAME = CASE_SENSITIVE;"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "id": "6c8b168c-c5c0-4af7-86f4-7ade69949283",
      "metadata": {
        "language": "sql",
        "name": "cell51"
      },
      "outputs": [],
      "source": [
        "ALTER WAREHOUSE HOL_WH SET WAREHOUSE_SIZE = XSMALL;"
      ]
    },
    {
      "cell_type": "markdown",
      "id": "5009d825-7703-4370-a26a-072aaf7a8a56",
      "metadata": {
        "name": "cell52"
      },
      "source": [
        "### Viewing the Task History\n",
        "Like the in the previous step, to see what happened when you ran this task DAG, highlight and run (using CMD/CTRL+Enter) this commented query in the script:\n",
        "\n",
        "```sql\n",
        "SELECT *\n",
        "FROM TABLE(INFORMATION_SCHEMA.TASK_HISTORY(\n",
        "    SCHEDULED_TIME_RANGE_START=>DATEADD('DAY',-1,CURRENT_TIMESTAMP()),\n",
        "    RESULT_LIMIT => 100))\n",
        "ORDER BY SCHEDULED_TIME DESC\n",
        ";\n",
        "```\n",
        "\n",
        "This time you will notice that the `ORDERS_UPDATE_TASK` task will not be skipped, since the `HARMONIZED.POS_FLATTENED_V_STREAM` stream has new data. In a few minutes you should see that both the `ORDERS_UPDATE_TASK` task and the `DAILY_CITY_METRICS_UPDATE_TASK` task completed successfully.\n",
        "\n",
        "### Query History for Tasks\n",
        "One important thing to understand about tasks, is that the queries which get executed by the task won't show up with the default Query History UI settings. In order to see the queries that just ran you need to do the following:\n",
        "\n",
        "* Remove filters at the top of this table, including your username, as later scheduled tasks will run as \"System\":\n",
        "\n",
        "![](assets/query_history_remove_filter1.png)\n",
        "\n",
        "* Click \"Filter\", and add filter option 'Queries executed by user tasks' and click \"Apply Filters\":\n",
        "\n",
        "![](assets/query_history_remove_filter2.png)\n",
        "\n",
        "You should now see all the queries run by your tasks! Take a look at each of the MERGE commands in the Query History to see how many records were processed by each task. And don't forget to notice that we processed the whole pipeline just now, and did so incrementally!\n"
      ]
    },
    {
      "cell_type": "markdown",
      "id": "ef47cbae-164e-48b2-8ab0-fafb6b4b09ab",
      "metadata": {
        "collapsed": false,
        "jupyter": {
          "outputs_hidden": false
        },
        "name": "cell53"
      },
      "source": [
        "## Teardown\n",
        "Once you're finished with the Quickstart and want to clean things up, you can simply run the following commands.\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "id": "bec5625e-83a3-405b-a78b-07d82088ef7e",
      "metadata": {
        "codeCollapsed": false,
        "collapsed": false,
        "language": "sql",
        "name": "cell54"
      },
      "outputs": [],
      "source": "-- DROP DATABASE NB_HOL_DB;\n-- DROP WAREHOUSE HOL_WH;"
    },
    {
      "cell_type": "markdown",
      "id": "bb371043-ba33-4d70-bc49-e388162c6e82",
      "metadata": {
        "name": "cell55"
      },
      "source": [
        "### What we've covered\n",
        "We've covered a ton in this Quickstart, and here are the highlights:\n",
        "\n",
        "* Snowflake's Table Format\n",
        "* Data ingestion with COPY\n",
        "* Schema inference\n",
        "* Data sharing/marketplace (instead of ETL)\n",
        "* Streams for incremental processing (CDC)\n",
        "* Streams on views\n",
        "* Python UDFs (with third-party packages)\n",
        "* Python Stored Procedures\n",
        "* Snowpark DataFrame API\n",
        "* Snowpark Python programmability\n",
        "* Warehouse elasticity (dynamic scaling)\n",
        "* Visual Studio Code Snowflake native extension (PuPr, Git integration)\n",
        "* SnowCLI (PuPr)\n",
        "* Tasks (with Stream triggers)\n",
        "* Task Observability\n",
        "* GitHub Actions (CI/CD) integration\n",
        "\n",
        "### Related Resources\n",
        "And finally, here's a quick recap of related resources:\n",
        "\n",
        "* [Full Demo on Snowflake Demo Hub](https://developers.snowflake.com/demos/data-engineering-pipelines/)\n",
        "* [Source Code on GitHub](https://github.com/Snowflake-Labs/sfguide-data-engineering-with-snowpark-python)\n",
        "* [Snowpark Developer Guide for Python](https://docs.snowflake.com/en/developer-guide/snowpark/python/index.html)\n",
        "    * [Writing Python UDFs](https://docs.snowflake.com/en/developer-guide/udf/python/udf-python.html)\n",
        "    * [Writing Stored Procedures in Snowpark (Python)](https://docs.snowflake.com/en/sql-reference/stored-procedures-python.html)\n",
        "    * [Working with DataFrames in Snowpark Python](https://docs.snowflake.com/en/developer-guide/snowpark/python/working-with-dataframes.html)\n",
        "* Related Tools\n",
        "    * [Snowflake Visual Studio Code Extension](https://marketplace.visualstudio.com/items?itemName=snowflake.snowflake-vsc)\n",
        "    * [SnowCLI Tool](https://github.com/Snowflake-Labs/snowcli)"
      ]
    }
  ],
  "metadata": {
    "kernelspec": {
      "display_name": "Python 3 (ipykernel)",
      "language": "python",
      "name": "python3"
    },
    "language_info": {
      "codemirror_mode": {
        "name": "ipython",
        "version": 3
      },
      "file_extension": ".py",
      "mimetype": "text/x-python",
      "name": "python",
      "nbconvert_exporter": "python",
      "pygments_lexer": "ipython3",
      "version": "3.11.5"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 4
}