notebooks/assignments/Week2Notebook.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Scope of Notebook"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The goal of this notebook is to showcase how you can prepare data for the future goal of consumption by an ML model, and leveraging functionality in the Adobe Experience Platform to generate features at scale and make it available in your choice of cloud storage.\n",
    "\n",
    "![Workflow](../media/CMLE-Notebooks-Week2-Workflow.png)\n",
    "\n",
    "We'll go through several steps:\n",
    "- **Creating a query** to encapsulate what a good featurized dataset will be.\n",
    "- **Executing that query** and storing the results.\n",
    "- **Setting up a flow** to export the results into cloud storage.\n",
    "- **Executing that flow** to deliver the results."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Setup"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This notebook requires some configuration data to properly authenticate to your Adobe Experience Platform instance. You should be able to find all the values required above by following the Setup section of the **README**.\n",
    "\n",
    "The next cell will be looking for your configuration file under your **ADOBE_HOME** path to fetch the values used throughout this notebook. See more details in the Setup section of the **README** to understand how to create your configuration file."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import getpass\n",
    "from configparser import ConfigParser\n",
    "\n",
    "if \"ADOBE_HOME\" not in os.environ:\n",
    "    raise Exception(\"ADOBE_HOME environment variable needs to be set.\")\n",
    "\n",
    "config = ConfigParser()\n",
    "config_path = os.path.join(os.environ[\"ADOBE_HOME\"], \"conf\", \"config.ini\")\n",
    "\n",
    "if not os.path.exists(config_path):\n",
    "    raise Exception(f\"Looking for configuration under {config_path} but config not found, please verify path\")\n",
    "\n",
    "config.read(config_path)\n",
    "\n",
    "ims_org_id = config.get(\"Platform\", \"ims_org_id\")\n",
    "sandbox_name = config.get(\"Platform\", \"sandbox_name\")\n",
    "environment = config.get(\"Platform\", \"environment\")\n",
    "client_id = config.get(\"Authentication\", \"client_id\")\n",
    "client_secret = config.get(\"Authentication\", \"client_secret\")\n",
    "scopes = config.get(\"Authentication\", \"scopes\")\n",
    "dataset_id = config.get(\"Platform\", \"dataset_id\")\n",
    "if not dataset_id:\n",
    "    raise Exception(\"Looking for dataset_id saved in week1 notebook to continue the exploration in week2 notebook but not found\")\n",
    "export_path = config.get(\"Cloud\", \"export_path\")\n",
    "data_format = config.get(\"Cloud\", \"data_format\")\n",
    "compression_type = config.get(\"Cloud\", \"compression_type\")\n",
    "username=getpass.getuser()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Some utility functions that will be used throughout this notebook:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_ui_link(tenant_id, resource_type, resource_id):\n",
    "    if environment == \"prod\":\n",
    "        prefix = f\"https://experience.adobe.com\"\n",
    "    else:\n",
    "        prefix = f\"https://experience-{environment}.adobe.com\"\n",
    "    return f\"{prefix}/#/@{tenant_id}/sname:{sandbox_name}/platform/{resource_type}/{resource_id}\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Before we run anything, make sure to install the following required libraries for this notebook. They are all publicly available libraries and the latest version should work fine.\n",
    "\n",
    "<div class=\"alert alert-block alert-warning\">\n",
    "<b>Note:</b> \n",
    "    \n",
    "If you are not using `pip` to manage Python packages, or that you use a variation like `pip3`, please update the cell below accordingly to install these packages.\n",
    "</div>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Requirement already satisfied: aepp in /usr/local/lib/python3.10/site-packages (0.2.7)\n",
      "Requirement already satisfied: PyJWT in /usr/local/lib/python3.10/site-packages (from aepp) (2.6.0)\n",
      "Requirement already satisfied: pathlib in /usr/local/lib/python3.10/site-packages (from aepp) (1.0.1)\n",
      "Requirement already satisfied: pathlib2 in /usr/local/lib/python3.10/site-packages (from aepp) (2.3.7.post1)\n",
      "Requirement already satisfied: requests in /usr/local/lib/python3.10/site-packages (from aepp) (2.28.2)\n",
      "Requirement already satisfied: pandas in /usr/local/lib/python3.10/site-packages (from aepp) (1.5.3)\n",
      "Requirement already satisfied: numpy>=1.21.0 in /usr/local/lib/python3.10/site-packages (from pandas->aepp) (1.24.2)\n",
      "Requirement already satisfied: pytz>=2020.1 in /usr/local/lib/python3.10/site-packages (from pandas->aepp) (2022.7.1)\n",
      "Requirement already satisfied: python-dateutil>=2.8.1 in /usr/local/lib/python3.10/site-packages (from pandas->aepp) (2.8.2)\n",
      "Requirement already satisfied: six in /usr/local/lib/python3.10/site-packages (from pathlib2->aepp) (1.16.0)\n",
      "Requirement already satisfied: cryptography>=3.4.0 in /usr/local/lib/python3.10/site-packages (from PyJWT->aepp) (39.0.2)\n",
      "Requirement already satisfied: certifi>=2017.4.17 in /usr/local/lib/python3.10/site-packages (from requests->aepp) (2022.12.7)\n",
      "Requirement already satisfied: charset-normalizer<4,>=2 in /usr/local/lib/python3.10/site-packages (from requests->aepp) (3.0.1)\n",
      "Requirement already satisfied: idna<4,>=2.5 in /usr/local/lib/python3.10/site-packages (from requests->aepp) (3.4)\n",
      "Requirement already satisfied: urllib3<1.27,>=1.21.1 in /usr/local/lib/python3.10/site-packages (from requests->aepp) (1.26.14)\n",
      "Requirement already satisfied: cffi>=1.12 in /usr/local/lib/python3.10/site-packages (from cryptography>=3.4.0->PyJWT->aepp) (1.15.1)\n",
      "Requirement already satisfied: pycparser in /usr/local/lib/python3.10/site-packages (from cffi>=1.12->cryptography>=3.4.0->PyJWT->aepp) (2.21)\n",
      "Requirement already satisfied: pygresql in /usr/local/lib/python3.10/site-packages (5.2.4)\n"
     ]
    }
   ],
   "source": [
    "!pip install \"aepp>=0.3.1-6\"\n",
    "!pip install pygresql==5.2.5"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We'll be using the [aepp Python library](https://github.com/pitchmuc/aepp) here to interact with AEP APIs and create a schema and dataset suitable for adding our synthetic data further down the line. This library simply provides a programmatic interface around the REST APIs, but all these steps could be completed similarly using the raw APIs directly or even in the UI. For more information on the underlying APIs please see [the API reference guide](https://developer.adobe.com/experience-platform-apis/).\n",
    "\n",
    "Before any calls can take place, we need to configure the library and setup authentication credentials. For this you'll need the following piece of information. For information about how you can get these, please refer to the `Setup` section of the **README**:\n",
    "- Client ID\n",
    "- Client secret"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "import aepp\n",
    "\n",
    "aepp.configure(\n",
    "  org_id=ims_org_id,\n",
    "  secret=client_secret,\n",
    "  scopes=scopes,\n",
    "  client_id=client_id,\n",
    "  sandbox=sandbox_name,\n",
    "  environment=environment\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 1. Creating Featurization Query"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In the previous week we created some synthetic data under a dataset in your Adobe Experience Platform instance, and now we're ready to use it to generate features that can then be fed to our ML model. For this purpose we'll be using the [Query Service](https://experienceleague.adobe.com/docs/experience-platform/query/home.html?lang=en#) which lets us access data from any dataset and run queries at scale. The end goal here is to compress this dataset into a small subset of meaningful features that will be relevant to our model."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Before we can issue queries, we need to find the table name corresponding to our dataset. Please make sure your dataset ID was entered in your configuration as part of the setup, it should be available under the `dataset_id` variable in this notebook.\n",
    "\n",
    "Every dataset in AEP should have a corresponding table in PQS based on its name. We can use AEP APIs to get that information with the code below:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'cmle_week1_new_notebook_with_ecid_used_dataset_created_by_saikat'"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from aepp import catalog\n",
    "\n",
    "cat_conn = catalog.Catalog()\n",
    "\n",
    "dataset_info = cat_conn.getDataSet(dataset_id)\n",
    "table_name = dataset_info[dataset_id][\"tags\"][\"adobe/pqs/table\"][0]\n",
    "table_name"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "And because some of the data we created in the previous week is under a custom field group, we need to fetch your tenant ID since the data will be nested under it. This can be accomplished simply with the code below"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'aamds'"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from aepp import schema\n",
    "\n",
    "schema_conn = schema.Schema()\n",
    "\n",
    "tenant_id = schema_conn.getTenantId()\n",
    "tenant_id"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we can use that table to query it via Query Service. Every query is able to run at scale leveraging Spark-based distributed computing power in the backend, so the goal is to take this large dataset, extract meaningful features and only keep a smaller subset to feed into a ML model.\n",
    "\n",
    "We'll be leveraging `aepp` again to interact with Query Service."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "from aepp import queryservice\n",
    "\n",
    "qs = queryservice.QueryService()\n",
    "qs_conn = qs.connection()\n",
    "qs_conn['dbname']=f'{sandbox_name}:{table_name}'\n",
    "qs_cursor = queryservice.InteractiveQuery(qs_conn)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<div class=\"alert alert-block alert-warning\">\n",
    "<b>Note:</b> If at any point in this notebook the connection to Query Service is closed, you can refresh it by re-enabling that cell.\n",
    "</div>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's first define our ML problem scientifically:\n",
    "- **What kind of problem** are we solving? We'd like to predict whether someone is likely to subscribe or not. We treat that as a binary classification problem, you are either subscribed or you are not.\n",
    "- What is our **target variable**? Whether a subscription occurred or not for a given user.\n",
    "- What are our **positive labels**? People who have at least 1 event corresponding to a subscription (marked as `web.formFilledOut`). We will keep a single feature row from the event where they actually subscribed.\n",
    "- What are our **negative labels**? People who don't have a single event corresponding to a subscription. We will keep a random row to avoid having bias in the data.\n",
    "\n",
    "Let's put that into practice and start looking at our positive labels:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Number of positive classes: 15030\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>eventType</th>\n",
       "      <th>userId</th>\n",
       "      <th>subscriptionOccurred</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>web.formFilledOut</td>\n",
       "      <td>01054714817856066632264746967668888198</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>web.formFilledOut</td>\n",
       "      <td>01108987448326166452882902459550847869</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>web.formFilledOut</td>\n",
       "      <td>01234932524049684301600430771576676651</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>web.formFilledOut</td>\n",
       "      <td>01242509742845154266705851412711275265</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>web.formFilledOut</td>\n",
       "      <td>01370262440242301894771514187123049829</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "           eventType                                  userId  \\\n",
       "0  web.formFilledOut  01054714817856066632264746967668888198   \n",
       "1  web.formFilledOut  01108987448326166452882902459550847869   \n",
       "2  web.formFilledOut  01234932524049684301600430771576676651   \n",
       "3  web.formFilledOut  01242509742845154266705851412711275265   \n",
       "4  web.formFilledOut  01370262440242301894771514187123049829   \n",
       "\n",
       "   subscriptionOccurred  \n",
       "0                     1  \n",
       "1                     1  \n",
       "2                     1  \n",
       "3                     1  \n",
       "4                     1  "
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "query_positive_labels = f\"\"\"\n",
    "SELECT *\n",
    "FROM (\n",
    "    SELECT\n",
    "        eventType,\n",
    "        _{tenant_id}.userid as userId,\n",
    "        SUM(CASE WHEN eventType='web.formFilledOut' THEN 1 ELSE 0 END) \n",
    "            OVER (PARTITION BY _{tenant_id}.userid) \n",
    "            AS \"subscriptionOccurred\"\n",
    "    FROM {table_name}\n",
    ")\n",
    "WHERE subscriptionOccurred = 1 AND eventType = 'web.formFilledOut'\n",
    "\"\"\"\n",
    "\n",
    "df_positive_labels = qs_cursor.query(query_positive_labels, output=\"dataframe\")\n",
    "print(f\"Number of positive classes: {len(df_positive_labels)}\")\n",
    "df_positive_labels.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<div class=\"alert alert-block alert-warning\">\n",
    "<b>Note:</b> \n",
    "    \n",
    "We are using a sub-query because we want to filter on the `subscriptionOccurred` which is defined as part of the query, so it can't be used in a filter condition unless it is in a sub-query.\n",
    "</div>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now let's look at our negative labels. Because we just want to retain a random row to avoid bias, we need to introduce randomness into our query:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Number of negative classes: 50000\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>userId</th>\n",
       "      <th>subscriptionOccurred</th>\n",
       "      <th>random_row_number_for_user</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>01027994177972439148069092698714414382</td>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>01066367784525869370502097766291450913</td>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>01117296890525140996735553609305695042</td>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>01149554820363915324573708359099551093</td>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>01172121447143590196349410086995740317</td>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                                   userId  subscriptionOccurred  \\\n",
       "0  01027994177972439148069092698714414382                     0   \n",
       "1  01066367784525869370502097766291450913                     0   \n",
       "2  01117296890525140996735553609305695042                     0   \n",
       "3  01149554820363915324573708359099551093                     0   \n",
       "4  01172121447143590196349410086995740317                     0   \n",
       "\n",
       "   random_row_number_for_user  \n",
       "0                           1  \n",
       "1                           1  \n",
       "2                           1  \n",
       "3                           1  \n",
       "4                           1  "
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "query_negative_labels = f\"\"\"\n",
    "SELECT *\n",
    "FROM (\n",
    "    SELECT\n",
    "        _{tenant_id}.userid as userId,\n",
    "        SUM(CASE WHEN eventType='web.formFilledOut' THEN 1 ELSE 0 END) \n",
    "            OVER (PARTITION BY _{tenant_id}.userid) \n",
    "            AS \"subscriptionOccurred\",\n",
    "        row_number() OVER (PARTITION BY _{tenant_id}.userid ORDER BY randn()) AS random_row_number_for_user\n",
    "    FROM {table_name}\n",
    ")\n",
    "WHERE subscriptionOccurred = 0 AND random_row_number_for_user = 1\n",
    "\"\"\"\n",
    "\n",
    "df_negative_labels = qs_cursor.query(query_negative_labels, output=\"dataframe\")\n",
    "print(f\"Number of negative classes: {len(df_negative_labels)}\")\n",
    "df_negative_labels.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Putting it all together, we can query both our positive and negative classes with the following query:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Number of classes: 50000\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>eventType</th>\n",
       "      <th>userId</th>\n",
       "      <th>subscriptionOccurred</th>\n",
       "      <th>random_row_number_for_user</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>commerce.productViews</td>\n",
       "      <td>01027994177972439148069092698714414382</td>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>web.formFilledOut</td>\n",
       "      <td>01054714817856066632264746967668888198</td>\n",
       "      <td>1</td>\n",
       "      <td>5</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>web.formFilledOut</td>\n",
       "      <td>01108987448326166452882902459550847869</td>\n",
       "      <td>1</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>directMarketing.emailSent</td>\n",
       "      <td>01117296890525140996735553609305695042</td>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>directMarketing.emailOpened</td>\n",
       "      <td>01149554820363915324573708359099551093</td>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                     eventType                                  userId  \\\n",
       "0        commerce.productViews  01027994177972439148069092698714414382   \n",
       "1            web.formFilledOut  01054714817856066632264746967668888198   \n",
       "2            web.formFilledOut  01108987448326166452882902459550847869   \n",
       "3    directMarketing.emailSent  01117296890525140996735553609305695042   \n",
       "4  directMarketing.emailOpened  01149554820363915324573708359099551093   \n",
       "\n",
       "   subscriptionOccurred  random_row_number_for_user  \n",
       "0                     0                           1  \n",
       "1                     1                           5  \n",
       "2                     1                           1  \n",
       "3                     0                           1  \n",
       "4                     0                           1  "
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "query_labels = f\"\"\"\n",
    "SELECT *\n",
    "FROM (\n",
    "    SELECT\n",
    "        eventType,\n",
    "        _{tenant_id}.userid as userId,\n",
    "        SUM(CASE WHEN eventType='web.formFilledOut' THEN 1 ELSE 0 END) \n",
    "            OVER (PARTITION BY _{tenant_id}.userid) \n",
    "            AS \"subscriptionOccurred\",\n",
    "        row_number() OVER (PARTITION BY _{tenant_id}.userid ORDER BY randn()) AS random_row_number_for_user\n",
    "    FROM {table_name}\n",
    ")\n",
    "WHERE (subscriptionOccurred = 1 AND eventType = 'web.formFilledOut') OR (subscriptionOccurred = 0 AND random_row_number_for_user = 1)\n",
    "\"\"\"\n",
    "\n",
    "df_labels = qs_cursor.query(query_labels, output=\"dataframe\")\n",
    "print(f\"Number of classes: {len(df_labels)}\")\n",
    "df_labels.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now let's think what kind of features make sense for this kind of problem that we would like to eventually feed to an ML model. There's 2 main kinds of features we're interested in:\n",
    "- **Actionable features**: different actions the user actually took in response to a marketing event.\n",
    "- **Temporal features**: distribution over time of the actions the user took. This is useful for modeling to capture behavior over time and not just statically at any particular point in time.\n",
    "\n",
    "We can think of a few simple things that can be derived from this data for the actionable features:\n",
    "- **Number of emails** that were sent for marketing purposes and received by the user.\n",
    "- Portion of these emails that were actually **opened**.\n",
    "- Portion of these emails where the user actually **clicked** on the link.\n",
    "- **Number of products** that were viewed.\n",
    "- Number of **propositions that were interacted with**.\n",
    "- Number of **propositions that were dismissed**.\n",
    "- Number of **links that were clicked on**.\n",
    "\n",
    "Regarding the temporal features, we can look at consecutive occurrences between various events. This can be accomplished using the `TIME_BETWEEN_PREVIOUS_MATCH` function. So we take the previous actionable features and look at their temporal distribution:\n",
    "- Number of minutes between 2 consecutive emails received.\n",
    "- Number of minutes between 2 consecutive emails opened.\n",
    "- Number of minutes between 2 consecutive emails where the user actually clicked on the link.\n",
    "- Number of minutes between 2 consecutive product views.\n",
    "- Number of minutes between 2 propositions that were interacted with.\n",
    "- Number of minutes between 2 propositions that were dismissed.\n",
    "- Number of minutes between 2 links that were clicked on.\n",
    "\n",
    "Let's put all that in practice and look how we can create these features inside a query:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>userId</th>\n",
       "      <th>emailsReceived</th>\n",
       "      <th>emailsOpened</th>\n",
       "      <th>emailsClicked</th>\n",
       "      <th>productsViewed</th>\n",
       "      <th>propositionInteracts</th>\n",
       "      <th>propositionDismissed</th>\n",
       "      <th>webLinkClicks</th>\n",
       "      <th>minutes_since_emailSent</th>\n",
       "      <th>minutes_since_emailOpened</th>\n",
       "      <th>minutes_since_emailClick</th>\n",
       "      <th>minutes_since_productView</th>\n",
       "      <th>minutes_since_propositionInteract</th>\n",
       "      <th>minutes_since_propositionDismiss</th>\n",
       "      <th>minutes_since_linkClick</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>01046464403641925015932682550398209571</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>None</td>\n",
       "      <td>NaN</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>01046464403641925015932682550398209571</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>None</td>\n",
       "      <td>NaN</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>01046464403641925015932682550398209571</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>None</td>\n",
       "      <td>NaN</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>01046464403641925015932682550398209571</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>None</td>\n",
       "      <td>0.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>01046464403641925015932682550398209571</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>None</td>\n",
       "      <td>1.0</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                                   userId  emailsReceived  emailsOpened  \\\n",
       "0  01046464403641925015932682550398209571               0             0   \n",
       "1  01046464403641925015932682550398209571               0             0   \n",
       "2  01046464403641925015932682550398209571               0             0   \n",
       "3  01046464403641925015932682550398209571               0             0   \n",
       "4  01046464403641925015932682550398209571               0             0   \n",
       "\n",
       "   emailsClicked  productsViewed  propositionInteracts  propositionDismissed  \\\n",
       "0              0               0                     0                     0   \n",
       "1              0               0                     0                     0   \n",
       "2              0               0                     0                     0   \n",
       "3              0               0                     0                     0   \n",
       "4              0               0                     0                     0   \n",
       "\n",
       "   webLinkClicks  minutes_since_emailSent  minutes_since_emailOpened  \\\n",
       "0              0                      NaN                        NaN   \n",
       "1              0                      NaN                        NaN   \n",
       "2              0                      NaN                        NaN   \n",
       "3              1                      NaN                        NaN   \n",
       "4              1                      NaN                        NaN   \n",
       "\n",
       "   minutes_since_emailClick  minutes_since_productView  \\\n",
       "0                       NaN                        NaN   \n",
       "1                       NaN                        NaN   \n",
       "2                       NaN                        NaN   \n",
       "3                       NaN                        NaN   \n",
       "4                       NaN                        NaN   \n",
       "\n",
       "   minutes_since_propositionInteract minutes_since_propositionDismiss  \\\n",
       "0                                NaN                             None   \n",
       "1                                NaN                             None   \n",
       "2                                NaN                             None   \n",
       "3                                NaN                             None   \n",
       "4                                NaN                             None   \n",
       "\n",
       "   minutes_since_linkClick  \n",
       "0                      NaN  \n",
       "1                      NaN  \n",
       "2                      NaN  \n",
       "3                      0.0  \n",
       "4                      1.0  "
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "query_features = f\"\"\"\n",
    "SELECT\n",
    "    _{tenant_id}.userid as userId,\n",
    "    SUM(CASE WHEN eventType='directMarketing.emailSent' THEN 1 ELSE 0 END) \n",
    "       OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "       AS \"emailsReceived\",\n",
    "    SUM(CASE WHEN eventType='directMarketing.emailOpened' THEN 1 ELSE 0 END) \n",
    "       OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "       AS \"emailsOpened\",       \n",
    "    SUM(CASE WHEN eventType='directMarketing.emailClicked' THEN 1 ELSE 0 END) \n",
    "       OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "       AS \"emailsClicked\",       \n",
    "    SUM(CASE WHEN eventType='commerce.productViews' THEN 1 ELSE 0 END) \n",
    "       OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "       AS \"productsViewed\",       \n",
    "    SUM(CASE WHEN eventType='decisioning.propositionInteract' THEN 1 ELSE 0 END) \n",
    "       OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "       AS \"propositionInteracts\",       \n",
    "    SUM(CASE WHEN eventType='decisioning.propositionDismiss' THEN 1 ELSE 0 END) \n",
    "       OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "       AS \"propositionDismissed\",\n",
    "    SUM(CASE WHEN eventType='web.webinteraction.linkClicks' THEN 1 ELSE 0 END) \n",
    "       OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "       AS \"webLinkClicks\" ,\n",
    "    TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'directMarketing.emailSent', 'minutes')\n",
    "       OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "       AS \"minutes_since_emailSent\",\n",
    "    TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'directMarketing.emailOpened', 'minutes')\n",
    "       OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "       AS \"minutes_since_emailOpened\",\n",
    "    TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'directMarketing.emailClicked', 'minutes')\n",
    "       OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "       AS \"minutes_since_emailClick\",\n",
    "    TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'commerce.productViews', 'minutes')\n",
    "       OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "       AS \"minutes_since_productView\",\n",
    "    TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'decisioning.propositionInteract', 'minutes')\n",
    "       OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "       AS \"minutes_since_propositionInteract\",\n",
    "    TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'propositionDismiss', 'minutes')\n",
    "       OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "       AS \"minutes_since_propositionDismiss\",\n",
    "    TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'web.webinteraction.linkClicks', 'minutes')\n",
    "       OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "       AS \"minutes_since_linkClick\"\n",
    "FROM {table_name}\n",
    "\"\"\"\n",
    "\n",
    "df_features = qs_cursor.query(query_features, output=\"dataframe\")\n",
    "df_features.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "At that point we have defined all our features, and we also have our classes cleanly defined, so we can tie everything together in a final query that will represent our training set to be used later on on our ML model."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>userId</th>\n",
       "      <th>eventType</th>\n",
       "      <th>timestamp</th>\n",
       "      <th>subscriptionOccurred</th>\n",
       "      <th>emailsReceived</th>\n",
       "      <th>emailsOpened</th>\n",
       "      <th>emailsClicked</th>\n",
       "      <th>productsViewed</th>\n",
       "      <th>propositionInteracts</th>\n",
       "      <th>propositionDismissed</th>\n",
       "      <th>webLinkClicks</th>\n",
       "      <th>minutes_since_emailSent</th>\n",
       "      <th>minutes_since_emailOpened</th>\n",
       "      <th>minutes_since_emailClick</th>\n",
       "      <th>minutes_since_productView</th>\n",
       "      <th>minutes_since_propositionInteract</th>\n",
       "      <th>minutes_since_propositionDismiss</th>\n",
       "      <th>minutes_since_linkClick</th>\n",
       "      <th>random_row_number_for_user</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>01024918979328021024268398155289293841</td>\n",
       "      <td>advertising.impressions</td>\n",
       "      <td>2023-01-11 11:01:28.844</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>None</td>\n",
       "      <td>NaN</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>01313428821394797049107854074567613407</td>\n",
       "      <td>directMarketing.emailSent</td>\n",
       "      <td>2023-01-11 11:07:12.511</td>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>None</td>\n",
       "      <td>NaN</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>02993793241626222851527839833265197633</td>\n",
       "      <td>web.formFilledOut</td>\n",
       "      <td>2023-01-11 15:36:21.667</td>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>1.0</td>\n",
       "      <td>None</td>\n",
       "      <td>NaN</td>\n",
       "      <td>10</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>03284744679335392483733573117571851574</td>\n",
       "      <td>advertising.impressions</td>\n",
       "      <td>2023-01-11 19:53:50.963</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>None</td>\n",
       "      <td>NaN</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>03077253005436651532049293706745952081</td>\n",
       "      <td>directMarketing.emailOpened</td>\n",
       "      <td>2023-01-11 20:45:29.123</td>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "      <td>3</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>829.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>None</td>\n",
       "      <td>NaN</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                                   userId                    eventType  \\\n",
       "0  01024918979328021024268398155289293841      advertising.impressions   \n",
       "1  01313428821394797049107854074567613407    directMarketing.emailSent   \n",
       "2  02993793241626222851527839833265197633            web.formFilledOut   \n",
       "3  03284744679335392483733573117571851574      advertising.impressions   \n",
       "4  03077253005436651532049293706745952081  directMarketing.emailOpened   \n",
       "\n",
       "                 timestamp  subscriptionOccurred  emailsReceived  \\\n",
       "0  2023-01-11 11:01:28.844                     0               0   \n",
       "1  2023-01-11 11:07:12.511                     0               1   \n",
       "2  2023-01-11 15:36:21.667                     1               0   \n",
       "3  2023-01-11 19:53:50.963                     0               0   \n",
       "4  2023-01-11 20:45:29.123                     0               1   \n",
       "\n",
       "   emailsOpened  emailsClicked  productsViewed  propositionInteracts  \\\n",
       "0             0              0               0                     0   \n",
       "1             0              0               0                     0   \n",
       "2             0              0               0                     1   \n",
       "3             0              0               0                     0   \n",
       "4             3              0               0                     0   \n",
       "\n",
       "   propositionDismissed  webLinkClicks  minutes_since_emailSent  \\\n",
       "0                     0              0                      NaN   \n",
       "1                     0              0                      0.0   \n",
       "2                     0              0                      NaN   \n",
       "3                     0              0                      NaN   \n",
       "4                     0              0                    829.0   \n",
       "\n",
       "   minutes_since_emailOpened  minutes_since_emailClick  \\\n",
       "0                        NaN                       NaN   \n",
       "1                        NaN                       NaN   \n",
       "2                        NaN                       NaN   \n",
       "3                        NaN                       NaN   \n",
       "4                        0.0                       NaN   \n",
       "\n",
       "   minutes_since_productView  minutes_since_propositionInteract  \\\n",
       "0                        NaN                                NaN   \n",
       "1                        NaN                                NaN   \n",
       "2                        NaN                                1.0   \n",
       "3                        NaN                                NaN   \n",
       "4                        NaN                                NaN   \n",
       "\n",
       "  minutes_since_propositionDismiss  minutes_since_linkClick  \\\n",
       "0                             None                      NaN   \n",
       "1                             None                      NaN   \n",
       "2                             None                      NaN   \n",
       "3                             None                      NaN   \n",
       "4                             None                      NaN   \n",
       "\n",
       "   random_row_number_for_user  \n",
       "0                           1  \n",
       "1                           1  \n",
       "2                          10  \n",
       "3                           1  \n",
       "4                           1  "
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "query_training_set = f\"\"\"\n",
    "SELECT *\n",
    "FROM (\n",
    "    SELECT _{tenant_id}.userid as userId, \n",
    "       eventType,\n",
    "       timestamp,\n",
    "       SUM(CASE WHEN eventType='web.formFilledOut' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid) \n",
    "           AS \"subscriptionOccurred\",\n",
    "       SUM(CASE WHEN eventType='directMarketing.emailSent' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"emailsReceived\",\n",
    "       SUM(CASE WHEN eventType='directMarketing.emailOpened' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"emailsOpened\",       \n",
    "       SUM(CASE WHEN eventType='directMarketing.emailClicked' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"emailsClicked\",       \n",
    "       SUM(CASE WHEN eventType='commerce.productViews' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"productsViewed\",       \n",
    "       SUM(CASE WHEN eventType='decisioning.propositionInteract' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"propositionInteracts\",       \n",
    "       SUM(CASE WHEN eventType='decisioning.propositionDismiss' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"propositionDismissed\",\n",
    "       SUM(CASE WHEN eventType='web.webinteraction.linkClicks' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"webLinkClicks\" ,\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'directMarketing.emailSent', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_emailSent\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'directMarketing.emailOpened', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_emailOpened\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'directMarketing.emailClicked', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_emailClick\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'commerce.productViews', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_productView\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'decisioning.propositionInteract', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_propositionInteract\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'propositionDismiss', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_propositionDismiss\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'web.webinteraction.linkClicks', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_linkClick\",\n",
    "        row_number() OVER (PARTITION BY _{tenant_id}.userid ORDER BY randn()) AS random_row_number_for_user\n",
    "    FROM {table_name} LIMIT 1000\n",
    ")\n",
    "WHERE (subscriptionOccurred = 1 AND eventType = 'web.formFilledOut') OR (subscriptionOccurred = 0 AND random_row_number_for_user = 1)\n",
    "ORDER BY timestamp\n",
    "\"\"\"\n",
    "\n",
    "df_training_set = qs_cursor.query(query_training_set, output=\"dataframe\")\n",
    "df_training_set.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 2. Generating Features Incrementally"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now in a typical ML workload you'll want to use incremental data to feed to your model, or data between some specific dates. For that purpose we can use snapshot information that is tracked inside Query Service every time a new batch of data is ingested, using the `history_meta` metadata table. For example, you can access the metadata for each batch of your dataset using the query below:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Total number of snapshots/batches: 10\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>snapshot_generation</th>\n",
       "      <th>made_current_at</th>\n",
       "      <th>snapshot_id</th>\n",
       "      <th>parent_id</th>\n",
       "      <th>is_current_ancestor</th>\n",
       "      <th>is_current</th>\n",
       "      <th>output_record_count</th>\n",
       "      <th>output_byte_size</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>0</td>\n",
       "      <td>2023-04-04 15:44:44.955</td>\n",
       "      <td>3825256663640337857</td>\n",
       "      <td>NaN</td>\n",
       "      <td>True</td>\n",
       "      <td>False</td>\n",
       "      <td>106317</td>\n",
       "      <td>14641809</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>1</td>\n",
       "      <td>2023-04-04 15:44:46.256</td>\n",
       "      <td>5250805769364495228</td>\n",
       "      <td>3.825257e+18</td>\n",
       "      <td>True</td>\n",
       "      <td>False</td>\n",
       "      <td>106736</td>\n",
       "      <td>14741300</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>2</td>\n",
       "      <td>2023-04-04 15:44:48.604</td>\n",
       "      <td>8316771932049439874</td>\n",
       "      <td>5.250806e+18</td>\n",
       "      <td>True</td>\n",
       "      <td>False</td>\n",
       "      <td>106059</td>\n",
       "      <td>14683324</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>3</td>\n",
       "      <td>2023-04-04 15:44:50.096</td>\n",
       "      <td>7241083319934178488</td>\n",
       "      <td>8.316772e+18</td>\n",
       "      <td>True</td>\n",
       "      <td>False</td>\n",
       "      <td>106474</td>\n",
       "      <td>14697877</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>4</td>\n",
       "      <td>2023-04-04 15:44:52.253</td>\n",
       "      <td>3157125847948566759</td>\n",
       "      <td>7.241083e+18</td>\n",
       "      <td>True</td>\n",
       "      <td>False</td>\n",
       "      <td>106675</td>\n",
       "      <td>14725877</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5</th>\n",
       "      <td>5</td>\n",
       "      <td>2023-04-04 15:44:59.995</td>\n",
       "      <td>4460330323826197864</td>\n",
       "      <td>3.157126e+18</td>\n",
       "      <td>True</td>\n",
       "      <td>False</td>\n",
       "      <td>106716</td>\n",
       "      <td>14734401</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>6</th>\n",
       "      <td>6</td>\n",
       "      <td>2023-04-04 15:45:19.237</td>\n",
       "      <td>4818579672832050765</td>\n",
       "      <td>4.460330e+18</td>\n",
       "      <td>True</td>\n",
       "      <td>False</td>\n",
       "      <td>106525</td>\n",
       "      <td>14711442</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>7</th>\n",
       "      <td>7</td>\n",
       "      <td>2023-04-04 15:45:31.781</td>\n",
       "      <td>7870632534228575321</td>\n",
       "      <td>4.818580e+18</td>\n",
       "      <td>True</td>\n",
       "      <td>False</td>\n",
       "      <td>106576</td>\n",
       "      <td>14750647</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>8</th>\n",
       "      <td>8</td>\n",
       "      <td>2023-04-04 15:45:43.676</td>\n",
       "      <td>1776236004746194930</td>\n",
       "      <td>7.870633e+18</td>\n",
       "      <td>True</td>\n",
       "      <td>False</td>\n",
       "      <td>107138</td>\n",
       "      <td>14766853</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>9</th>\n",
       "      <td>9</td>\n",
       "      <td>2023-04-04 15:45:44.813</td>\n",
       "      <td>6433588580290892078</td>\n",
       "      <td>1.776236e+18</td>\n",
       "      <td>True</td>\n",
       "      <td>True</td>\n",
       "      <td>107056</td>\n",
       "      <td>14779922</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   snapshot_generation          made_current_at          snapshot_id  \\\n",
       "0                    0  2023-04-04 15:44:44.955  3825256663640337857   \n",
       "1                    1  2023-04-04 15:44:46.256  5250805769364495228   \n",
       "2                    2  2023-04-04 15:44:48.604  8316771932049439874   \n",
       "3                    3  2023-04-04 15:44:50.096  7241083319934178488   \n",
       "4                    4  2023-04-04 15:44:52.253  3157125847948566759   \n",
       "5                    5  2023-04-04 15:44:59.995  4460330323826197864   \n",
       "6                    6  2023-04-04 15:45:19.237  4818579672832050765   \n",
       "7                    7  2023-04-04 15:45:31.781  7870632534228575321   \n",
       "8                    8  2023-04-04 15:45:43.676  1776236004746194930   \n",
       "9                    9  2023-04-04 15:45:44.813  6433588580290892078   \n",
       "\n",
       "      parent_id  is_current_ancestor  is_current  output_record_count  \\\n",
       "0           NaN                 True       False               106317   \n",
       "1  3.825257e+18                 True       False               106736   \n",
       "2  5.250806e+18                 True       False               106059   \n",
       "3  8.316772e+18                 True       False               106474   \n",
       "4  7.241083e+18                 True       False               106675   \n",
       "5  3.157126e+18                 True       False               106716   \n",
       "6  4.460330e+18                 True       False               106525   \n",
       "7  4.818580e+18                 True       False               106576   \n",
       "8  7.870633e+18                 True       False               107138   \n",
       "9  1.776236e+18                 True        True               107056   \n",
       "\n",
       "   output_byte_size  \n",
       "0          14641809  \n",
       "1          14741300  \n",
       "2          14683324  \n",
       "3          14697877  \n",
       "4          14725877  \n",
       "5          14734401  \n",
       "6          14711442  \n",
       "7          14750647  \n",
       "8          14766853  \n",
       "9          14779922  "
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "query_meta = f\"\"\"\n",
    "SELECT * FROM (SELECT history_meta('{table_name}'))\n",
    "\"\"\"\n",
    "\n",
    "df_meta = qs_cursor.query(query_meta, output=\"dataframe\")\n",
    "print(f\"Total number of snapshots/batches: {len(df_meta)}\")\n",
    "df_meta.head(n=10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now let's use that information to transform our featurization query into an incremental version of it. We can use [anonymous blocks](https://experienceleague.adobe.com/docs/experience-platform/query/sql/syntax.html?lang=en#anonymous-block) to create variables used to filter on the snapshots. Anonymous blocks are useful to embed multiple queries at once and do things like defining variables and such. We can then use Query Service's `SNAPSHOT BETWEEN x AND y` functionality to query data incrementally.\n",
    "\n",
    "In our case because this is the first time generating the features we will look at data between the most recent snapshot and its preceding one (which should correspond to the last batch of data ingested), but this can be extended to query data between any snapshots:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'\\n$$ BEGIN\\n\\nSET @from_snapshot_id = SELECT parent_id FROM (SELECT history_meta(\\'cmle_week1_new_notebook_with_ecid_used_dataset_created_by_saikat\\')) WHERE is_current = true;\\nSET @to_snapshot_id = SELECT snapshot_id FROM (SELECT history_meta(\\'cmle_week1_new_notebook_with_ecid_used_dataset_created_by_saikat\\')) WHERE is_current = true;\\n\\nSELECT *\\nFROM (\\n    SELECT _aamds.userid as userId, \\n       eventType,\\n       timestamp,\\n       SUM(CASE WHEN eventType=\\'web.formFilledOut\\' THEN 1 ELSE 0 END) \\n           OVER (PARTITION BY _aamds.userid) \\n           AS \"subscriptionOccurred\",\\n       SUM(CASE WHEN eventType=\\'directMarketing.emailSent\\' THEN 1 ELSE 0 END) \\n           OVER (PARTITION BY _aamds.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \\n           AS \"emailsReceived\",\\n       SUM(CASE WHEN eventType=\\'directMarketing.emailOpened\\' THEN 1 ELSE 0 END) \\n           OVER (PARTITION BY _aamds.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \\n           AS \"emailsOpened\",       \\n       SUM(CASE WHEN eventType=\\'directMarketing.emailClicked\\' THEN 1 ELSE 0 END) \\n           OVER (PARTITION BY _aamds.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \\n           AS \"emailsClicked\",       \\n       SUM(CASE WHEN eventType=\\'commerce.productViews\\' THEN 1 ELSE 0 END) \\n           OVER (PARTITION BY _aamds.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \\n           AS \"productsViewed\",       \\n       SUM(CASE WHEN eventType=\\'decisioning.propositionInteract\\' THEN 1 ELSE 0 END) \\n           OVER (PARTITION BY _aamds.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \\n           AS \"propositionInteracts\",       \\n       SUM(CASE WHEN eventType=\\'decisioning.propositionDismiss\\' THEN 1 ELSE 0 END) \\n           OVER (PARTITION BY _aamds.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \\n           AS \"propositionDismissed\",\\n       SUM(CASE WHEN eventType=\\'web.webinteraction.linkClicks\\' THEN 1 ELSE 0 END) \\n           OVER (PARTITION BY _aamds.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \\n           AS \"webLinkClicks\" ,\\n       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = \\'directMarketing.emailSent\\', \\'minutes\\')\\n           OVER (PARTITION BY _aamds.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \\n           AS \"minutes_since_emailSent\",\\n       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = \\'directMarketing.emailOpened\\', \\'minutes\\')\\n           OVER (PARTITION BY _aamds.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \\n           AS \"minutes_since_emailOpened\",\\n       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = \\'directMarketing.emailClicked\\', \\'minutes\\')\\n           OVER (PARTITION BY _aamds.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \\n           AS \"minutes_since_emailClick\",\\n       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = \\'commerce.productViews\\', \\'minutes\\')\\n           OVER (PARTITION BY _aamds.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \\n           AS \"minutes_since_productView\",\\n       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = \\'decisioning.propositionInteract\\', \\'minutes\\')\\n           OVER (PARTITION BY _aamds.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \\n           AS \"minutes_since_propositionInteract\",\\n       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = \\'propositionDismiss\\', \\'minutes\\')\\n           OVER (PARTITION BY _aamds.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \\n           AS \"minutes_since_propositionDismiss\",\\n       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = \\'web.webinteraction.linkClicks\\', \\'minutes\\')\\n           OVER (PARTITION BY _aamds.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \\n           AS \"minutes_since_linkClick\",\\n        row_number() OVER (PARTITION BY _aamds.userid ORDER BY randn()) AS random_row_number_for_user\\n    FROM cmle_week1_new_notebook_with_ecid_used_dataset_created_by_saikat\\n    SNAPSHOT BETWEEN @from_snapshot_id AND @to_snapshot_id\\n)\\nWHERE (subscriptionOccurred = 1 AND eventType = \\'web.formFilledOut\\') OR (subscriptionOccurred = 0 AND random_row_number_for_user = 1)\\nORDER BY timestamp;\\n\\nEXCEPTION\\n  WHEN OTHER THEN\\n    SELECT \\'ERROR\\';\\n\\nEND $$;\\n'"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "f\"\"\"\n",
    "$$ BEGIN\n",
    "\n",
    "SET @from_snapshot_id = SELECT parent_id FROM (SELECT history_meta('{table_name}')) WHERE is_current = true;\n",
    "SET @to_snapshot_id = SELECT snapshot_id FROM (SELECT history_meta('{table_name}')) WHERE is_current = true;\n",
    "\n",
    "SELECT *\n",
    "FROM (\n",
    "    SELECT _{tenant_id}.userid as userId, \n",
    "       eventType,\n",
    "       timestamp,\n",
    "       SUM(CASE WHEN eventType='web.formFilledOut' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid) \n",
    "           AS \"subscriptionOccurred\",\n",
    "       SUM(CASE WHEN eventType='directMarketing.emailSent' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"emailsReceived\",\n",
    "       SUM(CASE WHEN eventType='directMarketing.emailOpened' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"emailsOpened\",       \n",
    "       SUM(CASE WHEN eventType='directMarketing.emailClicked' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"emailsClicked\",       \n",
    "       SUM(CASE WHEN eventType='commerce.productViews' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"productsViewed\",       \n",
    "       SUM(CASE WHEN eventType='decisioning.propositionInteract' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"propositionInteracts\",       \n",
    "       SUM(CASE WHEN eventType='decisioning.propositionDismiss' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"propositionDismissed\",\n",
    "       SUM(CASE WHEN eventType='web.webinteraction.linkClicks' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"webLinkClicks\" ,\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'directMarketing.emailSent', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_emailSent\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'directMarketing.emailOpened', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_emailOpened\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'directMarketing.emailClicked', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_emailClick\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'commerce.productViews', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_productView\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'decisioning.propositionInteract', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_propositionInteract\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'propositionDismiss', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_propositionDismiss\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'web.webinteraction.linkClicks', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_linkClick\",\n",
    "        row_number() OVER (PARTITION BY _{tenant_id}.userid ORDER BY randn()) AS random_row_number_for_user\n",
    "    FROM {table_name}\n",
    "    SNAPSHOT BETWEEN @from_snapshot_id AND @to_snapshot_id\n",
    ")\n",
    "WHERE (subscriptionOccurred = 1 AND eventType = 'web.formFilledOut') OR (subscriptionOccurred = 0 AND random_row_number_for_user = 1)\n",
    "ORDER BY timestamp;\n",
    "\n",
    "EXCEPTION\n",
    "  WHEN OTHER THEN\n",
    "    SELECT 'ERROR';\n",
    "\n",
    "END $$;\n",
    "\"\"\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Note that we're not executing it interactively because this anonymous block is actually multiple queries chained together, and there is no simple way to return multiple result sets using PostgreSQL libraries.\n",
    "\n",
    "To solve that we can executed the query asynchronously and add a `CREATE TABLE x AS` statement at the beginning of our featurization query, to then look in that table. Note that because this is executed asynchronously, it goes into the Query Service scheduler and will take a few minutes to start executing, unlike the code we've been running until now which was synchronous and instant."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Query started successfully and got assigned ID dc46636d-5c19-40f7-a772-524cc238a93e - it will take some time to execute\n"
     ]
    }
   ],
   "source": [
    "import time\n",
    "import sys\n",
    "\n",
    "ctas_table_name = f\"cmle_example_training_set_incremental_{username}\"\n",
    "\n",
    "query_training_set_incremental = f\"\"\"\n",
    "$$ BEGIN\n",
    "\n",
    "SET @from_snapshot_id = SELECT parent_id FROM (SELECT history_meta('{table_name}')) WHERE is_current = true;\n",
    "SET @to_snapshot_id = SELECT snapshot_id FROM (SELECT history_meta('{table_name}')) WHERE is_current = true;\n",
    "\n",
    "CREATE TABLE {ctas_table_name} AS\n",
    "SELECT *\n",
    "FROM (\n",
    "    SELECT _{tenant_id}.userid as userId, \n",
    "       eventType,\n",
    "       timestamp,\n",
    "       SUM(CASE WHEN eventType='web.formFilledOut' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid) \n",
    "           AS \"subscriptionOccurred\",\n",
    "       SUM(CASE WHEN eventType='directMarketing.emailSent' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"emailsReceived\",\n",
    "       SUM(CASE WHEN eventType='directMarketing.emailOpened' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"emailsOpened\",       \n",
    "       SUM(CASE WHEN eventType='directMarketing.emailClicked' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"emailsClicked\",       \n",
    "       SUM(CASE WHEN eventType='commerce.productViews' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"productsViewed\",       \n",
    "       SUM(CASE WHEN eventType='decisioning.propositionInteract' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"propositionInteracts\",       \n",
    "       SUM(CASE WHEN eventType='decisioning.propositionDismiss' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"propositionDismissed\",\n",
    "       SUM(CASE WHEN eventType='web.webinteraction.linkClicks' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"webLinkClicks\" ,\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'directMarketing.emailSent', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_emailSent\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'directMarketing.emailOpened', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_emailOpened\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'directMarketing.emailClicked', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_emailClick\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'commerce.productViews', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_productView\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'decisioning.propositionInteract', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_propositionInteract\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'propositionDismiss', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_propositionDismiss\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'web.webinteraction.linkClicks', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_linkClick\",\n",
    "        row_number() OVER (PARTITION BY _{tenant_id}.userid ORDER BY randn()) AS random_row_number_for_user\n",
    "    FROM {table_name}\n",
    "    SNAPSHOT BETWEEN @from_snapshot_id AND @to_snapshot_id LIMIT 1000\n",
    ")\n",
    "WHERE (subscriptionOccurred = 1 AND eventType = 'web.formFilledOut') OR (subscriptionOccurred = 0 AND random_row_number_for_user = 1)\n",
    "ORDER BY timestamp;\n",
    "\n",
    "EXCEPTION\n",
    "  WHEN OTHER THEN\n",
    "    SELECT 'ERROR';\n",
    "\n",
    "END $$;\n",
    "\"\"\"\n",
    "\n",
    "query_incremental_res = qs.postQueries(\n",
    "    name=\"[CMLE][Week2] Query to generate incremental training data\",\n",
    "    sql=query_training_set_incremental,\n",
    "    dbname=f\"{sandbox_name}:{table_name}\"\n",
    ")\n",
    "query_incremental_id = query_incremental_res[\"id\"]\n",
    "print(f\"Query started successfully and got assigned ID {query_incremental_id} - it will take some time to execute\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "def wait_for_query_completion(query_id):\n",
    "    while True:\n",
    "        query_info = qs.getQuery(query_id)\n",
    "        query_state = query_info[\"state\"]\n",
    "        if query_state in [\"SUCCESS\", \"FAILED\"]:\n",
    "            break\n",
    "        print(\"Query is still in progress, sleeping...\")\n",
    "        time.sleep(60)\n",
    "\n",
    "    duration_secs = query_info[\"elapsedTime\"] / 1000\n",
    "    if query_state == \"SUCCESS\":\n",
    "        print(f\"Query completed successfully in {duration_secs} seconds\")\n",
    "    else:\n",
    "        print(f\"Query failed with the following errors:\", file=sys.stderr)\n",
    "        for error in query_info[\"errors\"]:\n",
    "            print(f\"Error code {error['code']}: {error['message']}\", file=sys.stderr)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Query is still in progress, sleeping...\n",
      "Query is still in progress, sleeping...\n",
      "Query is still in progress, sleeping...\n",
      "Query is still in progress, sleeping...\n",
      "Query is still in progress, sleeping...\n",
      "Query is still in progress, sleeping...\n",
      "Query is still in progress, sleeping...\n",
      "Query completed successfully in 386.37 seconds\n"
     ]
    }
   ],
   "source": [
    "wait_for_query_completion(query_incremental_id)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<div class=\"alert alert-block alert-warning\">\n",
    "<b>Note:</b> This should run in less than 10 minutes. If for whatever reason this does not finish in that time frame, it may get stuck creating the batch if the ingestion service is busy ingesting other data in your organization. We advise waiting longer or reaching out to your Adobe contact if this does not complete.\n",
    "    \n",
    "The same comment applies to subsequent cells where we use `wait_for_query_completion`.\n",
    "</div>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This `CREATE TABLE x AS` statement actually does several steps:\n",
    "- It will create a brand **new dataset** in your Adobe Experience Platform organization and sandbox.\n",
    "- The **schema** for this dataset will be created ad-hoc to **match the fields** of our featurization query, so there is no need to manually create schemas and fieldgroups for this.\n",
    "\n",
    "You can verify that by going in the UI at the link below, and making sure it you see the dataset as shown in the screenshot further down. The number of records should correspond to the batch size you used in the previous week since we are querying a single snapshot/batch.\n",
    "\n",
    "<div class=\"alert alert-block alert-warning\">\n",
    "<b>Note:</b> Re-executing the same query will fail, because it will try to create a table and not insert into it. We're solving this problem in the next section.\n",
    "</div>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Dataset available as dataset ID 642d8e01ec08531c065b48b2 under https://experience.adobe.com/#/@aamds/sname:cmle/platform/dataset/browse/642d8e01ec08531c065b48b2\n"
     ]
    }
   ],
   "source": [
    "datasets_res = cat_conn.getDataSets(name=ctas_table_name)\n",
    "if len(datasets_res) != 1:\n",
    "    raise Exception(f\"Expected a single dataset but got {len(datasets_res)} ones\")\n",
    "ctas_dataset_id = list(datasets_res.keys())[0]\n",
    "ctas_dataset_link = get_ui_link(tenant_id, \"dataset/browse\", ctas_dataset_id)\n",
    "print(f\"Dataset available as dataset ID {ctas_dataset_id} under {ctas_dataset_link}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![CTAS](../media/CMLE-Notebooks-Week2-CTAS.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we can just query it to see the structure of the data and verify it matches our query:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>userId</th>\n",
       "      <th>eventType</th>\n",
       "      <th>timestamp</th>\n",
       "      <th>subscriptionOccurred</th>\n",
       "      <th>emailsReceived</th>\n",
       "      <th>emailsOpened</th>\n",
       "      <th>emailsClicked</th>\n",
       "      <th>productsViewed</th>\n",
       "      <th>propositionInteracts</th>\n",
       "      <th>propositionDismissed</th>\n",
       "      <th>webLinkClicks</th>\n",
       "      <th>minutes_since_emailSent</th>\n",
       "      <th>minutes_since_emailOpened</th>\n",
       "      <th>minutes_since_emailClick</th>\n",
       "      <th>minutes_since_productView</th>\n",
       "      <th>minutes_since_propositionInteract</th>\n",
       "      <th>minutes_since_propositionDismiss</th>\n",
       "      <th>minutes_since_linkClick</th>\n",
       "      <th>random_row_number_for_user</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>03496398227344712440123581041340192961</td>\n",
       "      <td>web.webpagedetails.pageViews</td>\n",
       "      <td>2023-01-10 14:44:38.161</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>NaN</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>01306047604382086256919152660635616402</td>\n",
       "      <td>directMarketing.emailSent</td>\n",
       "      <td>2023-01-10 23:28:31.161</td>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>NaN</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>01899181931701313368403223253046030321</td>\n",
       "      <td>directMarketing.emailSent</td>\n",
       "      <td>2023-01-12 09:54:47.161</td>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "      <td>0.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>404.0</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>03077253005436651532049293706745952081</td>\n",
       "      <td>directMarketing.emailSent</td>\n",
       "      <td>2023-01-12 10:15:25.161</td>\n",
       "      <td>0</td>\n",
       "      <td>2</td>\n",
       "      <td>3</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>809.0</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>NaN</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>03622335536366276762200832126146857535</td>\n",
       "      <td>directMarketing.emailSent</td>\n",
       "      <td>2023-01-12 23:25:00.161</td>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>NaN</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                                   userId                     eventType  \\\n",
       "0  03496398227344712440123581041340192961  web.webpagedetails.pageViews   \n",
       "1  01306047604382086256919152660635616402     directMarketing.emailSent   \n",
       "2  01899181931701313368403223253046030321     directMarketing.emailSent   \n",
       "3  03077253005436651532049293706745952081     directMarketing.emailSent   \n",
       "4  03622335536366276762200832126146857535     directMarketing.emailSent   \n",
       "\n",
       "                 timestamp  subscriptionOccurred  emailsReceived  \\\n",
       "0  2023-01-10 14:44:38.161                     0               0   \n",
       "1  2023-01-10 23:28:31.161                     0               1   \n",
       "2  2023-01-12 09:54:47.161                     0               1   \n",
       "3  2023-01-12 10:15:25.161                     0               2   \n",
       "4  2023-01-12 23:25:00.161                     0               1   \n",
       "\n",
       "   emailsOpened  emailsClicked  productsViewed  propositionInteracts  \\\n",
       "0             0              0               0                     0   \n",
       "1             0              0               0                     0   \n",
       "2             0              0               0                     0   \n",
       "3             3              0               0                     0   \n",
       "4             0              0               0                     0   \n",
       "\n",
       "   propositionDismissed  webLinkClicks  minutes_since_emailSent  \\\n",
       "0                     0              0                      NaN   \n",
       "1                     0              0                      0.0   \n",
       "2                     0              1                      0.0   \n",
       "3                     0              0                      0.0   \n",
       "4                     0              0                      0.0   \n",
       "\n",
       "   minutes_since_emailOpened minutes_since_emailClick  \\\n",
       "0                        NaN                     None   \n",
       "1                        NaN                     None   \n",
       "2                        NaN                     None   \n",
       "3                      809.0                     None   \n",
       "4                        NaN                     None   \n",
       "\n",
       "  minutes_since_productView minutes_since_propositionInteract  \\\n",
       "0                      None                              None   \n",
       "1                      None                              None   \n",
       "2                      None                              None   \n",
       "3                      None                              None   \n",
       "4                      None                              None   \n",
       "\n",
       "  minutes_since_propositionDismiss  minutes_since_linkClick  \\\n",
       "0                             None                      NaN   \n",
       "1                             None                      NaN   \n",
       "2                             None                    404.0   \n",
       "3                             None                      NaN   \n",
       "4                             None                      NaN   \n",
       "\n",
       "   random_row_number_for_user  \n",
       "0                           1  \n",
       "1                           1  \n",
       "2                           1  \n",
       "3                           1  \n",
       "4                           1  "
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "query_ctas = f\"\"\"\n",
    "SELECT * FROM {ctas_table_name} LIMIT 10;\n",
    "\"\"\"\n",
    "qs = queryservice.QueryService()\n",
    "qs_conn = qs.connection()\n",
    "qs_conn['dbname']=f'{sandbox_name}:{ctas_table_name}'\n",
    "qs_cursor = queryservice.InteractiveQuery(qs_conn)\n",
    "df_ctas = qs_cursor.query(query_ctas, output=\"dataframe\")\n",
    "df_ctas.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 3. Templatizing the Featurization Query"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we've got a complete featurization query that can also be used to generate features incrementally, but we still want to go further:\n",
    "- The **snapshot window should be configurable** and easy to change without having to constantly create new queries.\n",
    "- The query itself should be **stored in a templatized way** so it can be referred to easily.\n",
    "- The query should be able to **create the table** automatically as well as **inserting into** a pre-existing table.\n",
    "\n",
    "Query Service has this concept of [templates](https://experienceleague.adobe.com/docs/experience-platform/query/ui/query-templates.html?lang=en) that we will be leveraging in this section to satisfy the requirements mentioned above."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The first step is to make make sure we either create the table if it does not exist, otherwise insert into it. This can be done by checking if the table exists using the `table_exists` function and adding a condition in our anonymous block based on that:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Query started successfully and got assigned ID f7fd9bd5-cc7d-4984-b33a-f639cd48fd18 - it will take some time to execute\n",
      "Query is still in progress, sleeping...\n",
      "Query is still in progress, sleeping...\n",
      "Query is still in progress, sleeping...\n",
      "Query is still in progress, sleeping...\n",
      "Query completed successfully in 202.07 seconds\n"
     ]
    }
   ],
   "source": [
    "query_training_set_ctas_or_insert = f\"\"\"\n",
    "$$ BEGIN\n",
    "\n",
    "SET @from_snapshot_id = SELECT parent_id FROM (SELECT history_meta('{table_name}')) WHERE is_current = true;\n",
    "SET @to_snapshot_id = SELECT snapshot_id FROM (SELECT history_meta('{table_name}')) WHERE is_current = true;\n",
    "SET @my_table_exists = SELECT table_exists('{ctas_table_name}');\n",
    "\n",
    "CREATE TABLE IF NOT EXISTS {ctas_table_name} AS\n",
    "SELECT *\n",
    "FROM (\n",
    "    SELECT _{tenant_id}.userid as userId, \n",
    "       eventType,\n",
    "       timestamp,\n",
    "       SUM(CASE WHEN eventType='web.formFilledOut' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid) \n",
    "           AS \"subscriptionOccurred\",\n",
    "       SUM(CASE WHEN eventType='directMarketing.emailSent' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"emailsReceived\",\n",
    "       SUM(CASE WHEN eventType='directMarketing.emailOpened' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"emailsOpened\",       \n",
    "       SUM(CASE WHEN eventType='directMarketing.emailClicked' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"emailsClicked\",       \n",
    "       SUM(CASE WHEN eventType='commerce.productViews' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"productsViewed\",       \n",
    "       SUM(CASE WHEN eventType='decisioning.propositionInteract' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"propositionInteracts\",       \n",
    "       SUM(CASE WHEN eventType='decisioning.propositionDismiss' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"propositionDismissed\",\n",
    "       SUM(CASE WHEN eventType='web.webinteraction.linkClicks' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"webLinkClicks\" ,\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'directMarketing.emailSent', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_emailSent\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'directMarketing.emailOpened', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_emailOpened\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'directMarketing.emailClicked', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_emailClick\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'commerce.productViews', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_productView\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'decisioning.propositionInteract', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_propositionInteract\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'propositionDismiss', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_propositionDismiss\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'web.webinteraction.linkClicks', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_linkClick\",\n",
    "        row_number() OVER (PARTITION BY _{tenant_id}.userid ORDER BY randn()) AS random_row_number_for_user\n",
    "    FROM {table_name}\n",
    "    SNAPSHOT BETWEEN @from_snapshot_id AND @to_snapshot_id LIMIT 1000\n",
    ")\n",
    "WHERE (subscriptionOccurred = 1 AND eventType = 'web.formFilledOut') OR (subscriptionOccurred = 0 AND random_row_number_for_user = 1)\n",
    "ORDER BY timestamp;\n",
    "\n",
    "INSERT INTO {ctas_table_name}\n",
    "SELECT *\n",
    "FROM (\n",
    "    SELECT _{tenant_id}.userid as userId, \n",
    "       eventType,\n",
    "       timestamp,\n",
    "       SUM(CASE WHEN eventType='web.formFilledOut' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid) \n",
    "           AS \"subscriptionOccurred\",\n",
    "       SUM(CASE WHEN eventType='directMarketing.emailSent' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"emailsReceived\",\n",
    "       SUM(CASE WHEN eventType='directMarketing.emailOpened' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"emailsOpened\",       \n",
    "       SUM(CASE WHEN eventType='directMarketing.emailClicked' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"emailsClicked\",       \n",
    "       SUM(CASE WHEN eventType='commerce.productViews' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"productsViewed\",       \n",
    "       SUM(CASE WHEN eventType='decisioning.propositionInteract' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"propositionInteracts\",       \n",
    "       SUM(CASE WHEN eventType='decisioning.propositionDismiss' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"propositionDismissed\",\n",
    "       SUM(CASE WHEN eventType='web.webinteraction.linkClicks' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"webLinkClicks\" ,\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'directMarketing.emailSent', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_emailSent\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'directMarketing.emailOpened', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_emailOpened\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'directMarketing.emailClicked', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_emailClick\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'commerce.productViews', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_productView\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'decisioning.propositionInteract', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_propositionInteract\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'propositionDismiss', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_propositionDismiss\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'web.webinteraction.linkClicks', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_linkClick\",\n",
    "        row_number() OVER (PARTITION BY _{tenant_id}.userid ORDER BY randn()) AS random_row_number_for_user\n",
    "    FROM {table_name}\n",
    "    SNAPSHOT BETWEEN @from_snapshot_id AND @to_snapshot_id LIMIT 1000\n",
    ")\n",
    "WHERE \n",
    "    @my_table_exists = 't' AND\n",
    "    ((subscriptionOccurred = 1 AND eventType = 'web.formFilledOut') OR (subscriptionOccurred = 0 AND random_row_number_for_user = 1))\n",
    "ORDER BY timestamp;\n",
    "\n",
    "EXCEPTION\n",
    "  WHEN OTHER THEN\n",
    "    SELECT 'ERROR';\n",
    "\n",
    "END $$;\n",
    "\"\"\"\n",
    "\n",
    "query_ctas_or_insert_res = qs.postQueries(\n",
    "    name=\"[CMLE][Week2] Query to generate training data as CTAS or Insert\",\n",
    "    sql=query_training_set_ctas_or_insert,\n",
    "    dbname=f\"{sandbox_name}:all\"\n",
    ")\n",
    "query_ctas_or_insert_id = query_ctas_or_insert_res[\"id\"]\n",
    "print(f\"Query started successfully and got assigned ID {query_ctas_or_insert_id} - it will take some time to execute\")\n",
    "\n",
    "wait_for_query_completion(query_ctas_or_insert_id)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The next step is to make the snapshot time window configurable. To do that we can replace the part containing the snapshot boundaries with variables as `$variable` so they can be passed at runtime using Query Service:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "ctas_table_name = f\"cmle_training_set_{username}\"\n",
    "\n",
    "query_training_set_template = f\"\"\"\n",
    "$$ BEGIN\n",
    "\n",
    "SET @my_table_exists = SELECT table_exists('{ctas_table_name}');\n",
    "\n",
    "CREATE TABLE IF NOT EXISTS {ctas_table_name} AS\n",
    "SELECT *\n",
    "FROM (\n",
    "    SELECT _{tenant_id}.userid as userId, \n",
    "       eventType,\n",
    "       timestamp,\n",
    "       SUM(CASE WHEN eventType='web.formFilledOut' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid) \n",
    "           AS \"subscriptionOccurred\",\n",
    "       SUM(CASE WHEN eventType='directMarketing.emailSent' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"emailsReceived\",\n",
    "       SUM(CASE WHEN eventType='directMarketing.emailOpened' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"emailsOpened\",       \n",
    "       SUM(CASE WHEN eventType='directMarketing.emailClicked' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"emailsClicked\",       \n",
    "       SUM(CASE WHEN eventType='commerce.productViews' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"productsViewed\",       \n",
    "       SUM(CASE WHEN eventType='decisioning.propositionInteract' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"propositionInteracts\",       \n",
    "       SUM(CASE WHEN eventType='decisioning.propositionDismiss' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"propositionDismissed\",\n",
    "       SUM(CASE WHEN eventType='web.webinteraction.linkClicks' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"webLinkClicks\" ,\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'directMarketing.emailSent', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_emailSent\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'directMarketing.emailOpened', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_emailOpened\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'directMarketing.emailClicked', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_emailClick\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'commerce.productViews', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_productView\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'decisioning.propositionInteract', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_propositionInteract\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'propositionDismiss', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_propositionDismiss\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'web.webinteraction.linkClicks', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_linkClick\",\n",
    "        row_number() OVER (PARTITION BY _{tenant_id}.userid ORDER BY randn()) AS random_row_number_for_user\n",
    "    FROM {table_name}\n",
    "    SNAPSHOT BETWEEN $from_snapshot_id AND $to_snapshot_id\n",
    ")\n",
    "WHERE (subscriptionOccurred = 1 AND eventType = 'web.formFilledOut') OR (subscriptionOccurred = 0 AND random_row_number_for_user = 1)\n",
    "ORDER BY timestamp;\n",
    "\n",
    "INSERT INTO {ctas_table_name}\n",
    "SELECT *\n",
    "FROM (\n",
    "    SELECT _{tenant_id}.userid as userId, \n",
    "       eventType,\n",
    "       timestamp,\n",
    "       SUM(CASE WHEN eventType='web.formFilledOut' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid) \n",
    "           AS \"subscriptionOccurred\",\n",
    "       SUM(CASE WHEN eventType='directMarketing.emailSent' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"emailsReceived\",\n",
    "       SUM(CASE WHEN eventType='directMarketing.emailOpened' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"emailsOpened\",       \n",
    "       SUM(CASE WHEN eventType='directMarketing.emailClicked' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"emailsClicked\",       \n",
    "       SUM(CASE WHEN eventType='commerce.productViews' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"productsViewed\",       \n",
    "       SUM(CASE WHEN eventType='decisioning.propositionInteract' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"propositionInteracts\",       \n",
    "       SUM(CASE WHEN eventType='decisioning.propositionDismiss' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"propositionDismissed\",\n",
    "       SUM(CASE WHEN eventType='web.webinteraction.linkClicks' THEN 1 ELSE 0 END) \n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"webLinkClicks\" ,\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'directMarketing.emailSent', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_emailSent\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'directMarketing.emailOpened', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_emailOpened\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'directMarketing.emailClicked', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_emailClick\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'commerce.productViews', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_productView\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'decisioning.propositionInteract', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_propositionInteract\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'propositionDismiss', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_propositionDismiss\",\n",
    "       TIME_BETWEEN_PREVIOUS_MATCH(timestamp, eventType = 'web.webinteraction.linkClicks', 'minutes')\n",
    "           OVER (PARTITION BY _{tenant_id}.userid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) \n",
    "           AS \"minutes_since_linkClick\",\n",
    "        row_number() OVER (PARTITION BY _{tenant_id}.userid ORDER BY randn()) AS random_row_number_for_user\n",
    "    FROM {table_name}\n",
    "    SNAPSHOT BETWEEN $from_snapshot_id AND $to_snapshot_id\n",
    ")\n",
    "WHERE \n",
    "    @my_table_exists = 't' AND\n",
    "    ((subscriptionOccurred = 1 AND eventType = 'web.formFilledOut') OR (subscriptionOccurred = 0 AND random_row_number_for_user = 1))\n",
    "ORDER BY timestamp;\n",
    "\n",
    "EXCEPTION\n",
    "  WHEN OTHER THEN\n",
    "    SELECT 'ERROR';\n",
    "\n",
    "END $$;\n",
    "\"\"\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We're not executing it because it has actual variables in it that will need to be resolved at runtime, so executing it right now would fail. We're ready to turn this into a proper template, which requires the following:\n",
    "- A **name** for your templatized query.\n",
    "- Some set of **query parameters** that you might want to already save - in our case we're not setting any so both snapshot boundaries can be set at runtie.\n",
    "- Your SQL **query**.\n",
    "\n",
    "Once you do this, the template should be available in the UI at the link below, as you can see in the screenshot."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Query template for training data created as ID f1d7db3a-adcb-472b-ac7d-c91b355b1182 under https://experience.adobe.com/#/@aamds/sname:cmle/platform/query/edit/f1d7db3a-adcb-472b-ac7d-c91b355b1182\n"
     ]
    }
   ],
   "source": [
    "template_res = qs.createQueryTemplate({\n",
    "  \"sql\": query_training_set_template,\n",
    "  \"queryParameters\": {},\n",
    "  \"name\": f\"[CMLE][Week2] Template for training data created by {username}\"\n",
    "})\n",
    "template_id = template_res[\"id\"]\n",
    "template_link = get_ui_link(tenant_id, \"query/edit\", template_id)\n",
    "\n",
    "print(f\"Query template for training data created as ID {template_id} under {template_link}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![Template](../media/CMLE-Notebooks-Week2-Template.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now that the template is saved, we can refer to it at any time, and passing any kind of values for the snapshots that we want. So for example if you have streaming data coming through your system, you just need to find out the beginning snapshot ID and end snapshot ID, and you can execute this featurization query that will take care of querying between these 2 snapshots.\n",
    "\n",
    "In this example, we'll just query the entire dataset, so the very first and very last snapshots:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Query will go from start snapshot ID 3825256663640337857 to end snapshot ID 6433588580290892078\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>snapshot_id</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>3825256663640337857</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>6433588580290892078</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "           snapshot_id\n",
       "0  3825256663640337857\n",
       "1  6433588580290892078"
      ]
     },
     "execution_count": 22,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "query_snapshots = f\"\"\"\n",
    "SELECT snapshot_id \n",
    "FROM (\n",
    "    SELECT history_meta('{table_name}')\n",
    ") \n",
    "WHERE is_current = true OR snapshot_generation = 0 \n",
    "ORDER BY snapshot_generation ASC\n",
    "\"\"\"\n",
    "\n",
    "df_snapshots = qs_cursor.query(query_snapshots, output=\"dataframe\")\n",
    "\n",
    "snapshot_start_id = str(df_snapshots[\"snapshot_id\"].iloc[0])\n",
    "snapshot_end_id = str(df_snapshots[\"snapshot_id\"].iloc[1])\n",
    "print(f\"Query will go from start snapshot ID {snapshot_start_id} to end snapshot ID {snapshot_end_id}\")\n",
    "\n",
    "df_snapshots.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Query started successfully and got assigned ID cf117097-703c-4851-957b-8d8571119248 - it will take some time to execute\n",
      "Query is still in progress, sleeping...\n",
      "Query is still in progress, sleeping...\n",
      "Query is still in progress, sleeping...\n",
      "Query is still in progress, sleeping...\n",
      "Query is still in progress, sleeping...\n",
      "Query completed successfully in 284.399 seconds\n"
     ]
    }
   ],
   "source": [
    "query_final_res = qs.postQueries(\n",
    "    name=f\"[CMLE][Week2] Query to generate training data created by {username}\",\n",
    "    templateId=template_id,\n",
    "    queryParameters={\n",
    "        \"from_snapshot_id\": snapshot_start_id,\n",
    "        \"to_snapshot_id\": snapshot_end_id,\n",
    "    },\n",
    "    dbname=f\"{sandbox_name}:all\"\n",
    ")\n",
    "query_final_id = query_final_res[\"id\"]\n",
    "print(f\"Query started successfully and got assigned ID {query_final_id} - it will take some time to execute\")\n",
    "\n",
    "wait_for_query_completion(query_final_id)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "At that point we got our full featurized dataset that is ready to plug into a ML model. But it's still in an Adobe Experience Platform dataset so far, and we need to bring it back to our cloud storage account outside of the Experience Platform to use our tool of choice, which will be covered in the next section.\n",
    "\n",
    "Before we go through that, we just need to find te dataset ID corresponding to the output of our templatized query:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Featurized data available in table cmle_training_set_cmenguy under dataset ID 642d912512f86b1c07ef5650 under https://experience.adobe.com/#/@aamds/sname:cmle/platform/dataset/browse/642d912512f86b1c07ef5650\n"
     ]
    }
   ],
   "source": [
    "ctas_table_info = cat_conn.getDataSets(name=ctas_table_name)\n",
    "created_dataset_id = list(ctas_table_info.keys())[0]\n",
    "created_dataset_link = get_ui_link(tenant_id,\"dataset/browse\",created_dataset_id)\n",
    "print(f\"Featurized data available in table {ctas_table_name} under dataset ID {created_dataset_id} under {created_dataset_link}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 4. Exporting the Featurized Dataset to Cloud Storage"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now that our featurized data is in a dataset, we need to bring it out to an external cloud storage filesystem from which the ML model training and scoring will be performed. \n",
    "\n",
    "For the purposes of this notebook we will be using the [Data Landing Zone (DLZ)](https://experienceleague.adobe.com/docs/experience-platform/sources/api-tutorials/create/cloud-storage/data-landing-zone.html?lang=en) as the filesystem. Every Adobe Experience Platform has a DLZ already setup as an [Azure Blob Storage](https://azure.microsoft.com/en-us/products/storage/blobs) container. We'll be using that as a delivery mechanism for the featurized data, but this step can be customized to delivery this data to any cloud storage filesystem.\n",
    "\n",
    "To setup the delivery pipeline, we'll be using the [Flow Service for Destinations](https://developer.adobe.com/experience-platform-apis/references/destinations/) which will be responsible for picking up the featurized data and dump it into the DLZ. There's a few steps involved:\n",
    "- Creating a **base connection**.\n",
    "- Creating a **source connection**.\n",
    "- Creating a **target connection**.\n",
    "- Creating a **data flow**.\n",
    "- Creating a **flow run**.\n",
    "\n",
    "For that, again we use `aepp` to abstract all the APIs:"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 4.1 Exporting dataset to Data Landing Zone"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import logging\n",
    "from aepp import exportDatasetToDataLandingZone\n",
    "\n",
    "export = exportDatasetToDataLandingZone.ExportDatasetToDataLandingZone(loggingObject = {\n",
    "    \"level\": logging.INFO,\n",
    "    \"stream\": \"Y\",\n",
    "    \"format\": \"%(asctime)s;%(levelname)s;%(module)s;%(message)s\",\n",
    "    \"filename\":\"\",\n",
    "    \"file\":\"\"\n",
    "})\n",
    "\n",
    "dataflow_id = export.createDataFlowRunIfNotExists(\n",
    "    dataset_id = created_dataset_id,\n",
    "    data_format = data_format, \n",
    "    export_path= export_path, \n",
    "    compression_type = compression_type, \n",
    "    on_schedule = False, \n",
    "    config_path = config_path, \n",
    "    entity_name = f\"[CMLE][Week2] Flow for Featurized Dataset to DLZ created by {username}\"\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dataflow_link = get_ui_link(tenant_id, \"destination/browse\", dataflow_id)\n",
    "print(f\"Data Flow created as ID {dataflow_id} available under {dataflow_link}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![Dataflow](../media/CMLE-Notebooks-Week2-Dataflow.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now that a run of our Data Flow has executed successfully, we're all set! We can do a sanity check to verify that the data indeed made its way into the DLZ. For that, we recommend setting up [Azure Storage Explorer](https://azure.microsoft.com/en-us/products/storage/storage-explorer) to connect to your DLZ container using [this guide](https://experienceleague.adobe.com/docs/experience-platform/destinations/catalog/cloud-storage/data-landing-zone.html?lang=en). To get the credentials, you can execute the code below to get the SAS URL needed:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## 4.2 Connecting to Data Landing Zone Container"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "DLZ container: dlz-destination\n",
      "DLZ storage account: sndbxdtlndv43vd4kug0dnjn\n",
      "DLZ SAS URL: https://sndbxdtlndv43vd4kug0dnjn.blob.core.windows.net/dlz-destination?sv=2020-10-02&si=dlz-5fe94224-8440-428d-9d34-6c734cf147e4&sr=c&sp=rl&sig=mlsGbwD%2Fyy58IcD9wVHACFDRsW%2Fzv4AolJo5bw36Ofw%3D\n"
     ]
    }
   ],
   "source": [
    "# TODO: use functionality in aepp once released\n",
    "from aepp import connector\n",
    "\n",
    "connector = connector.AdobeRequest(\n",
    "  config_object=aepp.config.config_object,\n",
    "  header=aepp.config.header,\n",
    "  loggingEnabled=False,\n",
    "  logger=None,\n",
    ")\n",
    "\n",
    "endpoint = aepp.config.endpoints[\"global\"] + \"/data/foundation/connectors/landingzone/credentials\"\n",
    "\n",
    "dlz_credentials = connector.getData(endpoint=endpoint, params={\n",
    "  \"type\": \"dlz_destination\"\n",
    "})\n",
    "dlz_container = dlz_credentials[\"containerName\"]\n",
    "dlz_sas_token = dlz_credentials[\"SASToken\"]\n",
    "dlz_storage_account = dlz_credentials[\"storageAccountName\"]\n",
    "dlz_sas_uri = dlz_credentials[\"SASUri\"]\n",
    "print(f\"DLZ container: {dlz_container}\")\n",
    "print(f\"DLZ storage account: {dlz_storage_account}\")\n",
    "print(f\"DLZ SAS URL: {dlz_sas_uri}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Once setup you should be able to see your featurized data as a set of Parquet files under the following directory structure: `cmle/egress/$DATASETID/exportTime=$TIMESTAMP` - see screenshot below."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Featurized data in DLZ should be available under cmle/egress/641a8dc485c3d71b98f893df\n"
     ]
    }
   ],
   "source": [
    "print(f\"Featurized data in DLZ should be available under {export_path}/{created_dataset_id}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![DLZ](../media/CMLE-Notebooks-Week2-ExportedDataset.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 4.3 Saving the featurized dataset to the configuration"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now that we got everything working, we just need to save the `created_dataset_id` variable in the original configuration file, so we can refer to it in the following weekly assignments. To do that, execute the code below:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {},
   "outputs": [],
   "source": [
    "config.set(\"Platform\", \"featurized_dataset_id\", created_dataset_id)\n",
    "\n",
    "with open(config_path, \"w\") as configfile:\n",
    "    config.write(configfile)"
   ]
  }
 ],
 "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.12.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}