examples/end-to-end-session-based/01-ETL-with-NVTabular.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "cb8675ec",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Copyright 2022 NVIDIA Corporation. All Rights Reserved.\n",
    "#\n",
    "# Licensed under the Apache License, Version 2.0 (the \"License\");\n",
    "# you may not use this file except in compliance with the License.\n",
    "# You may obtain a copy of the License at\n",
    "#\n",
    "#     http://www.apache.org/licenses/LICENSE-2.0\n",
    "#\n",
    "# Unless required by applicable law or agreed to in writing, software\n",
    "# distributed under the License is distributed on an \"AS IS\" BASIS,\n",
    "# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n",
    "# See the License for the specific language governing permissions and\n",
    "# limitations under the License.\n",
    "# ==============================================================================\n",
    "\n",
    "# Each user is responsible for checking the content of datasets and the\n",
    "# applicable licenses and determining if suitable for the intended use."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4f5dc27e-daac-4529-9b76-826a4de4818a",
   "metadata": {},
   "source": [
    "<img src=\"https://developer.download.nvidia.com/notebooks/dlsw-notebooks/merlin_transformers4rec_end-to-end-session-based-01-etl-with-nvtabular/nvidia_logo.png\" style=\"width: 90px; float: right;\">\n",
    "\n",
    "# ETL with NVTabular\n",
    "\n",
    "This notebook is created using the latest stable [merlin-pytorch](https://catalog.ngc.nvidia.com/orgs/nvidia/teams/merlin/containers/merlin-pytorch) container.\n",
    "\n",
    "**Launch the docker container**\n",
    "```\n",
    "docker run -it --gpus device=0 -p 8000:8000 -p 8001:8001 -p 8002:8002 -p 8888:8888 -v <path_to_data>:/workspace/data/  nvcr.io/nvidia/merlin/merlin-pytorch:23.XX\n",
    "```\n",
    "This script will mount your local data folder that includes your data files to `/workspace/data` directory in the merlin-pytorch docker container."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "af23729e-c01e-4cd4-bd06-2e68ac5f61fb",
   "metadata": {},
   "source": [
    "## Overview"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f1e6d7c9",
   "metadata": {},
   "source": [
    "This notebook demonstrates how to use NVTabular to perform the feature engineering that is needed to model the `YOOCHOOSE` dataset which contains a collection of sessions from a retailer. Each session  encapsulates the click events that the user performed in that session.\n",
    "\n",
    "The dataset is available on [Kaggle](https://www.kaggle.com/chadgostopp/recsys-challenge-2015). You need to download it and copy to the `DATA_FOLDER` path. Note that we are only using the `yoochoose-clicks.dat` file.\n",
    "\n",
    "Alternatively, you can generate a synthetic dataset with the same columns and dtypes as the `YOOCHOOSE` dataset and a default date range of 5 days. If the environment variable `USE_SYNTHETIC` is set to `True`, the code below will execute the function `generate_synthetic_data` and the rest of the notebook will run on a synthetic dataset.\n",
    "\n",
    "First, let's start by importing several libraries:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "99fa1c8e",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/usr/local/lib/python3.8/dist-packages/merlin/dtypes/mappings/tf.py:52: UserWarning: Tensorflow dtype mappings did not load successfully due to an error: No module named 'tensorflow'\n",
      "  warn(f\"Tensorflow dtype mappings did not load successfully due to an error: {exc.msg}\")\n"
     ]
    }
   ],
   "source": [
    "import os\n",
    "import glob\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "import gc\n",
    "import calendar\n",
    "import datetime\n",
    "\n",
    "import cudf\n",
    "import cupy\n",
    "import nvtabular as nvt\n",
    "from merlin.dag import ColumnSelector\n",
    "from merlin.schema import Schema, Tags"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6ec27e86-546e-4857-88c1-03b46fc6de40",
   "metadata": {},
   "source": [
    "Avoid Numba low occupancy warnings:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "77687671-4442-4386-beb8-728abd3633b6",
   "metadata": {},
   "outputs": [],
   "source": [
    "from numba import config\n",
    "config.CUDA_LOW_OCCUPANCY_WARNINGS = 0"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d5c40762",
   "metadata": {},
   "source": [
    "#### Define Data Input and Output Paths"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "648a50e9",
   "metadata": {},
   "outputs": [],
   "source": [
    "DATA_FOLDER = os.environ.get(\"DATA_FOLDER\", \"/workspace/data\")\n",
    "FILENAME_PATTERN = 'yoochoose-clicks.dat'\n",
    "DATA_PATH = os.path.join(DATA_FOLDER, FILENAME_PATTERN)\n",
    "\n",
    "OUTPUT_FOLDER = \"./yoochoose_transformed\"\n",
    "OVERWRITE = False\n",
    "\n",
    "USE_SYNTHETIC = os.environ.get(\"USE_SYNTHETIC\", False)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f20e24e7",
   "metadata": {},
   "source": [
    "## Load and clean raw data"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3fba8546-668c-4743-960e-ea2aef99ef24",
   "metadata": {},
   "source": [
    "Execute the cell below if you would like to work with synthetic data. Otherwise you can skip and continue with the next cell."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "07d14289-c783-45f0-86e8-e5c1001bfd76",
   "metadata": {},
   "outputs": [],
   "source": [
    "def generate_synthetic_data(\n",
    "    start_date: datetime.date, end_date: datetime.date, rows_per_day: int = 10000\n",
    ") -> pd.DataFrame:\n",
    "    assert end_date > start_date, \"end_date must be later than start_date\"\n",
    "\n",
    "    number_of_days = (end_date - start_date).days\n",
    "    total_number_of_rows = number_of_days * rows_per_day\n",
    "\n",
    "    # Generate a long-tail distribution of item interactions. This simulates that some items are\n",
    "    # more popular than others.\n",
    "    long_tailed_item_distribution = np.clip(\n",
    "        np.random.lognormal(3.0, 1.0, total_number_of_rows).astype(np.int64), 1, 50000\n",
    "    )\n",
    "\n",
    "    # generate random item interaction features\n",
    "    df = pd.DataFrame(\n",
    "        {\n",
    "            \"session_id\": np.random.randint(70000, 80000, total_number_of_rows),\n",
    "            \"item_id\": long_tailed_item_distribution,\n",
    "        },\n",
    "    )\n",
    "\n",
    "    # generate category mapping for each item-id\n",
    "    df[\"category\"] = pd.cut(df[\"item_id\"], bins=334, labels=np.arange(1, 335)).astype(\n",
    "        np.int64\n",
    "    )\n",
    "\n",
    "    max_session_length = 60 * 60  # 1 hour\n",
    "\n",
    "    def add_timestamp_to_session(session: pd.DataFrame):\n",
    "        random_start_date_and_time = calendar.timegm(\n",
    "            (\n",
    "                start_date\n",
    "                # Add day offset from start_date\n",
    "                + datetime.timedelta(days=np.random.randint(0, number_of_days))\n",
    "                # Add time offset within the random day\n",
    "                + datetime.timedelta(seconds=np.random.randint(0, 86_400))\n",
    "            ).timetuple()\n",
    "        )\n",
    "        session[\"timestamp\"] = random_start_date_and_time + np.clip(\n",
    "            np.random.lognormal(3.0, 1.0, len(session)).astype(np.int64),\n",
    "            0,\n",
    "            max_session_length,\n",
    "        )\n",
    "        return session\n",
    "\n",
    "    df = df.groupby(\"session_id\").apply(add_timestamp_to_session).reset_index()\n",
    "\n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "f35dff52",
   "metadata": {},
   "outputs": [],
   "source": [
    "if USE_SYNTHETIC:\n",
    "    START_DATE = os.environ.get(\"START_DATE\", \"2014/4/1\")\n",
    "    END_DATE = os.environ.get(\"END_DATE\", \"2014/4/5\")\n",
    "    interactions_df = generate_synthetic_data(datetime.datetime.strptime(START_DATE, '%Y/%m/%d'),\n",
    "                                              datetime.datetime.strptime(END_DATE, '%Y/%m/%d'))\n",
    "    interactions_df = cudf.from_pandas(interactions_df)\n",
    "else:\n",
    "    interactions_df = cudf.read_csv(DATA_PATH, sep=',', \n",
    "                                    names=['session_id','timestamp', 'item_id', 'category'], \n",
    "                                    dtype=['int', 'datetime64[s]', 'int', 'int'])"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5209e41e",
   "metadata": {},
   "source": [
    "#### Remove repeated interactions within the same session"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "22c2df72",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Count with in-session repeated interactions: 33003944\n",
      "Count after removed in-session repeated interactions: 28971543\n"
     ]
    }
   ],
   "source": [
    "print(\"Count with in-session repeated interactions: {}\".format(len(interactions_df)))\n",
    "\n",
    "# Sorts the dataframe by session and timestamp, to remove consecutive repetitions\n",
    "interactions_df.timestamp = interactions_df.timestamp.astype(int)\n",
    "interactions_df = interactions_df.sort_values(['session_id', 'timestamp'])\n",
    "past_ids = interactions_df['item_id'].shift(1).fillna()\n",
    "session_past_ids = interactions_df['session_id'].shift(1).fillna()\n",
    "\n",
    "# Keeping only no consecutive repeated in session interactions\n",
    "interactions_df = interactions_df[~((interactions_df['session_id'] == session_past_ids) & (interactions_df['item_id'] == past_ids))]\n",
    "\n",
    "print(\"Count after removed in-session repeated interactions: {}\".format(len(interactions_df)))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5250e189",
   "metadata": {},
   "source": [
    "#### Create new feature with the timestamp when the item was first seen"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "66a1bd13",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "   session_id   timestamp    item_id  category  itemid_ts_first\n",
      "0        7401  1396439960  214826816         0       1396321828\n",
      "1        7402  1396780751  214613743         0       1396329089\n",
      "2        7402  1396780780  214827011         0       1396735848\n",
      "3        7402  1396780912  214821388         0       1396330458\n",
      "4        7402  1396780991  214827011         0       1396735848\n"
     ]
    }
   ],
   "source": [
    "items_first_ts_df = interactions_df.groupby('item_id').agg({'timestamp': 'min'}).reset_index().rename(columns={'timestamp': 'itemid_ts_first'})\n",
    "interactions_merged_df = interactions_df.merge(items_first_ts_df, on=['item_id'], how='left')\n",
    "print(interactions_merged_df.head())"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "719637f3",
   "metadata": {},
   "source": [
    "Let's save the interactions_merged_df to disk to be able to use in the inference step."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "a0f908a1",
   "metadata": {},
   "outputs": [],
   "source": [
    "if os.path.isdir(DATA_FOLDER) == False:\n",
    "    os.mkdir(DATA_FOLDER)\n",
    "interactions_merged_df.to_parquet(os.path.join(DATA_FOLDER, 'interactions_merged_df.parquet'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "909f87c5-bff5-48c8-b714-cc556a4bc64d",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "52739\n"
     ]
    }
   ],
   "source": [
    "# print the total number of unique items in the dataset\n",
    "print(interactions_merged_df.item_id.nunique())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "04a3b5b7",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "517"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# free gpu memory\n",
    "del interactions_df, session_past_ids, items_first_ts_df\n",
    "gc.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "be2991c6",
   "metadata": {},
   "source": [
    "##  Define a preprocessing workflow with NVTabular"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f2669af6",
   "metadata": {},
   "source": [
    "NVTabular is a feature engineering and preprocessing library for tabular data designed to quickly and easily manipulate terabyte scale datasets used to train deep learning based recommender systems. It provides a high level abstraction to simplify code and accelerates computation on the GPU using the RAPIDS cuDF library.\n",
    "\n",
    "NVTabular supports different feature engineering transformations required by deep learning (DL) models such as Categorical encoding and numerical feature normalization. It also supports feature engineering and generating sequential features. \n",
    "\n",
    "More information about the supported features can be found <a href=https://nvidia-merlin.github.io/NVTabular/> here. </a>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "69475ddc",
   "metadata": {},
   "source": [
    "### Feature engineering: Create and Transform items features"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e6de1f5e",
   "metadata": {},
   "source": [
    "In this cell, we are defining three transformations ops: \n",
    "\n",
    "- 1. Encoding categorical variables using `Categorify()` op. Categorify op maps nulls to `1`, OOVs to `2`, automatically. We reserve `0` for padding the sequence features. The encoding of each category starts from 3.\n",
    "- 2. Deriving temporal features from timestamp and computing their cyclical representation using a custom lambda function. \n",
    "- 3. Computing the item recency in days using a custom op. Note that item recency is defined as the difference between the first occurrence of the item in dataset and the actual date of item interaction. \n",
    "\n",
    "For more ETL workflow examples, visit NVTabular [example notebooks](https://github.com/NVIDIA-Merlin/NVTabular/tree/stable/examples)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "86f80068",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Encodes categorical features as contiguous integers\n",
    "cat_feats = ColumnSelector(['category', 'item_id']) >> nvt.ops.Categorify()\n",
    "\n",
    "# create time features\n",
    "session_ts = ColumnSelector(['timestamp'])\n",
    "session_time = (\n",
    "    session_ts >> \n",
    "    nvt.ops.LambdaOp(lambda col: cudf.to_datetime(col, unit='s')) >> \n",
    "    nvt.ops.Rename(name = 'event_time_dt')\n",
    ")\n",
    "sessiontime_weekday = (\n",
    "    session_time >> \n",
    "    nvt.ops.LambdaOp(lambda col: col.dt.weekday) >> \n",
    "    nvt.ops.Rename(name ='et_dayofweek')\n",
    ")\n",
    "\n",
    "# Derive cyclical features: Define a custom lambda function \n",
    "def get_cycled_feature_value_sin(col, max_value):\n",
    "    value_scaled = (col + 0.000001) / max_value\n",
    "    value_sin = np.sin(2*np.pi*value_scaled)\n",
    "    return value_sin\n",
    "\n",
    "weekday_sin = sessiontime_weekday >> (lambda col: get_cycled_feature_value_sin(col+1, 7)) >> nvt.ops.Rename(name = 'et_dayofweek_sin')\n",
    "\n",
    "# Compute Item recency: Define a custom Op \n",
    "class ItemRecency(nvt.ops.Operator):\n",
    "    def transform(self, columns, gdf):\n",
    "        for column in columns.names:\n",
    "            col = gdf[column]\n",
    "            item_first_timestamp = gdf['itemid_ts_first']\n",
    "            delta_days = (col - item_first_timestamp) / (60*60*24)\n",
    "            gdf[column + \"_age_days\"] = delta_days * (delta_days >=0)\n",
    "        return gdf\n",
    "\n",
    "    def compute_selector(\n",
    "        self,\n",
    "        input_schema: Schema,\n",
    "        selector: ColumnSelector,\n",
    "        parents_selector: ColumnSelector,\n",
    "        dependencies_selector: ColumnSelector,\n",
    "    ) -> ColumnSelector:\n",
    "        self._validate_matching_cols(input_schema, parents_selector, \"computing input selector\")\n",
    "        return parents_selector\n",
    "\n",
    "    def column_mapping(self, col_selector):\n",
    "        column_mapping = {}\n",
    "        for col_name in col_selector.names:\n",
    "            column_mapping[col_name + \"_age_days\"] = [col_name]\n",
    "        return column_mapping\n",
    "\n",
    "    @property\n",
    "    def dependencies(self):\n",
    "        return [\"itemid_ts_first\"]\n",
    "\n",
    "    @property\n",
    "    def output_dtype(self):\n",
    "        return np.float64\n",
    "    \n",
    "recency_features = session_ts >> ItemRecency() \n",
    "# Apply standardization to this continuous feature\n",
    "recency_features_norm = recency_features >> nvt.ops.LogOp() >> nvt.ops.Normalize(out_dtype=np.float32) >> nvt.ops.Rename(name='product_recency_days_log_norm')\n",
    "\n",
    "time_features = (\n",
    "    session_time +\n",
    "    sessiontime_weekday +\n",
    "    weekday_sin + \n",
    "    recency_features_norm\n",
    ")\n",
    "\n",
    "features = ColumnSelector(['session_id', 'timestamp']) + cat_feats + time_features "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "bb7dc1df",
   "metadata": {},
   "source": [
    "### Define the preprocessing of sequential features"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "018872e9",
   "metadata": {},
   "source": [
    "Once the item features are generated, the objective of this cell is to group interactions at the session level, sorting the interactions by time. We additionally truncate all sessions to first 20 interactions and filter out sessions with less than 2 interactions."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "10b5c96c",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Define Groupby Operator\n",
    "groupby_features = features >> nvt.ops.Groupby(\n",
    "    groupby_cols=[\"session_id\"], \n",
    "    sort_cols=[\"timestamp\"],\n",
    "    aggs={\n",
    "        'item_id': [\"list\", \"count\"],\n",
    "        'category': [\"list\"],  \n",
    "        'timestamp': [\"first\"],\n",
    "        'event_time_dt': [\"first\"],\n",
    "        'et_dayofweek_sin': [\"list\"],\n",
    "        'product_recency_days_log_norm': [\"list\"]\n",
    "        },\n",
    "    name_sep=\"-\")\n",
    "\n",
    "# Truncate sequence features to first interacted 20 items \n",
    "SESSIONS_MAX_LENGTH = 20 \n",
    "\n",
    "item_feat = groupby_features['item_id-list'] >> nvt.ops.TagAsItemID()\n",
    "cont_feats = groupby_features['et_dayofweek_sin-list', 'product_recency_days_log_norm-list'] >> nvt.ops.AddMetadata(tags=[Tags.CONTINUOUS])\n",
    "\n",
    "\n",
    "groupby_features_list =  item_feat + cont_feats + groupby_features['category-list']\n",
    "groupby_features_truncated = groupby_features_list >> nvt.ops.ListSlice(-SESSIONS_MAX_LENGTH)\n",
    "\n",
    "# Calculate session day index based on 'event_time_dt-first' column\n",
    "day_index = ((groupby_features['event_time_dt-first'])  >> \n",
    "             nvt.ops.LambdaOp(lambda col: (col - col.min()).dt.days +1) >> \n",
    "             nvt.ops.Rename(f = lambda col: \"day_index\") >>\n",
    "             nvt.ops.AddMetadata(tags=[Tags.CATEGORICAL])\n",
    "            )\n",
    "\n",
    "# tag session_id column for serving with legacy api\n",
    "sess_id = groupby_features['session_id'] >> nvt.ops.AddMetadata(tags=[Tags.CATEGORICAL])\n",
    "\n",
    "# Select features for training \n",
    "selected_features = sess_id + groupby_features['item_id-count'] + groupby_features_truncated + day_index\n",
    "\n",
    "# Filter out sessions with less than 2 interactions \n",
    "MINIMUM_SESSION_LENGTH = 2\n",
    "filtered_sessions = selected_features >> nvt.ops.Filter(f=lambda df: df[\"item_id-count\"] >= MINIMUM_SESSION_LENGTH) "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1465a499",
   "metadata": {},
   "source": [
    "### Execute NVTabular workflow"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "26fdbd0c",
   "metadata": {},
   "source": [
    "Once we have defined the general workflow (`filtered_sessions`), we provide our cudf dataset to `nvt.Dataset` class which is optimized to split data into chunks that can fit in device memory and to handle the calculation of complex global statistics. Then, we execute the pipeline that fits and transforms data to get the desired output features."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "45803886",
   "metadata": {},
   "outputs": [],
   "source": [
    "dataset = nvt.Dataset(interactions_merged_df)\n",
    "workflow = nvt.Workflow(filtered_sessions)\n",
    "# Learn features statistics necessary of the preprocessing workflow\n",
    "# The following will generate schema.pbtxt file in the provided folder and export the parquet files.\n",
    "workflow.fit_transform(dataset).to_parquet(os.path.join(DATA_FOLDER, \"processed_nvt\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fc8d0b7c",
   "metadata": {},
   "source": [
    "Let's print the head of our preprocessed dataset. You can notice that now each example (row) is a session and the sequential features with respect to user interactions were converted to lists with matching length."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "4c10efb5-89c5-4458-a634-475eb459a47c",
   "metadata": {
    "tags": []
   },
   "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>name</th>\n",
       "      <th>tags</th>\n",
       "      <th>dtype</th>\n",
       "      <th>is_list</th>\n",
       "      <th>is_ragged</th>\n",
       "      <th>properties.num_buckets</th>\n",
       "      <th>properties.freq_threshold</th>\n",
       "      <th>properties.max_size</th>\n",
       "      <th>properties.cat_path</th>\n",
       "      <th>properties.domain.min</th>\n",
       "      <th>properties.domain.max</th>\n",
       "      <th>properties.domain.name</th>\n",
       "      <th>properties.embedding_sizes.cardinality</th>\n",
       "      <th>properties.embedding_sizes.dimension</th>\n",
       "      <th>properties.value_count.min</th>\n",
       "      <th>properties.value_count.max</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>session_id</td>\n",
       "      <td>(Tags.CATEGORICAL)</td>\n",
       "      <td>DType(name='int64', element_type=&lt;ElementType....</td>\n",
       "      <td>False</td>\n",
       "      <td>False</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>item_id-count</td>\n",
       "      <td>(Tags.CATEGORICAL)</td>\n",
       "      <td>DType(name='int32', element_type=&lt;ElementType....</td>\n",
       "      <td>False</td>\n",
       "      <td>False</td>\n",
       "      <td>NaN</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>.//categories/unique.item_id.parquet</td>\n",
       "      <td>0.0</td>\n",
       "      <td>52741.0</td>\n",
       "      <td>item_id</td>\n",
       "      <td>52742.0</td>\n",
       "      <td>512.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>item_id-list</td>\n",
       "      <td>(Tags.CATEGORICAL, Tags.ID, Tags.LIST, Tags.ITEM)</td>\n",
       "      <td>DType(name='int64', element_type=&lt;ElementType....</td>\n",
       "      <td>True</td>\n",
       "      <td>True</td>\n",
       "      <td>NaN</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>.//categories/unique.item_id.parquet</td>\n",
       "      <td>0.0</td>\n",
       "      <td>52741.0</td>\n",
       "      <td>item_id</td>\n",
       "      <td>52742.0</td>\n",
       "      <td>512.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>20.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>et_dayofweek_sin-list</td>\n",
       "      <td>(Tags.CONTINUOUS, Tags.LIST)</td>\n",
       "      <td>DType(name='float64', element_type=&lt;ElementTyp...</td>\n",
       "      <td>True</td>\n",
       "      <td>True</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>0.0</td>\n",
       "      <td>20.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>product_recency_days_log_norm-list</td>\n",
       "      <td>(Tags.CONTINUOUS, Tags.LIST)</td>\n",
       "      <td>DType(name='float32', element_type=&lt;ElementTyp...</td>\n",
       "      <td>True</td>\n",
       "      <td>True</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>0.0</td>\n",
       "      <td>20.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5</th>\n",
       "      <td>category-list</td>\n",
       "      <td>(Tags.CATEGORICAL, Tags.LIST)</td>\n",
       "      <td>DType(name='int64', element_type=&lt;ElementType....</td>\n",
       "      <td>True</td>\n",
       "      <td>True</td>\n",
       "      <td>NaN</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>.//categories/unique.category.parquet</td>\n",
       "      <td>0.0</td>\n",
       "      <td>336.0</td>\n",
       "      <td>category</td>\n",
       "      <td>337.0</td>\n",
       "      <td>42.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>20.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>6</th>\n",
       "      <td>day_index</td>\n",
       "      <td>(Tags.CATEGORICAL)</td>\n",
       "      <td>DType(name='int64', element_type=&lt;ElementType....</td>\n",
       "      <td>False</td>\n",
       "      <td>False</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "[{'name': 'session_id', 'tags': {<Tags.CATEGORICAL: 'categorical'>}, 'properties': {}, 'dtype': DType(name='int64', element_type=<ElementType.Int: 'int'>, element_size=64, element_unit=None, signed=True, shape=Shape(dims=(Dimension(min=0, max=None),))), 'is_list': False, 'is_ragged': False}, {'name': 'item_id-count', 'tags': {<Tags.CATEGORICAL: 'categorical'>}, 'properties': {'num_buckets': None, 'freq_threshold': 0, 'max_size': 0, 'cat_path': './/categories/unique.item_id.parquet', 'domain': {'min': 0, 'max': 52741, 'name': 'item_id'}, 'embedding_sizes': {'cardinality': 52742, 'dimension': 512}}, 'dtype': DType(name='int32', element_type=<ElementType.Int: 'int'>, element_size=32, element_unit=None, signed=True, shape=Shape(dims=(Dimension(min=0, max=None),))), 'is_list': False, 'is_ragged': False}, {'name': 'item_id-list', 'tags': {<Tags.CATEGORICAL: 'categorical'>, <Tags.ID: 'id'>, <Tags.LIST: 'list'>, <Tags.ITEM: 'item'>}, 'properties': {'num_buckets': None, 'freq_threshold': 0, 'max_size': 0, 'cat_path': './/categories/unique.item_id.parquet', 'domain': {'min': 0, 'max': 52741, 'name': 'item_id'}, 'embedding_sizes': {'cardinality': 52742, 'dimension': 512}, 'value_count': {'min': 0, 'max': 20}}, 'dtype': DType(name='int64', element_type=<ElementType.Int: 'int'>, element_size=64, element_unit=None, signed=True, shape=Shape(dims=(Dimension(min=0, max=None), Dimension(min=0, max=20)))), 'is_list': True, 'is_ragged': True}, {'name': 'et_dayofweek_sin-list', 'tags': {<Tags.CONTINUOUS: 'continuous'>, <Tags.LIST: 'list'>}, 'properties': {'value_count': {'min': 0, 'max': 20}}, 'dtype': DType(name='float64', element_type=<ElementType.Float: 'float'>, element_size=64, element_unit=None, signed=True, shape=Shape(dims=(Dimension(min=0, max=None), Dimension(min=0, max=20)))), 'is_list': True, 'is_ragged': True}, {'name': 'product_recency_days_log_norm-list', 'tags': {<Tags.CONTINUOUS: 'continuous'>, <Tags.LIST: 'list'>}, 'properties': {'value_count': {'min': 0, 'max': 20}}, 'dtype': DType(name='float32', element_type=<ElementType.Float: 'float'>, element_size=32, element_unit=None, signed=True, shape=Shape(dims=(Dimension(min=0, max=None), Dimension(min=0, max=20)))), 'is_list': True, 'is_ragged': True}, {'name': 'category-list', 'tags': {<Tags.CATEGORICAL: 'categorical'>, <Tags.LIST: 'list'>}, 'properties': {'num_buckets': None, 'freq_threshold': 0, 'max_size': 0, 'cat_path': './/categories/unique.category.parquet', 'domain': {'min': 0, 'max': 336, 'name': 'category'}, 'embedding_sizes': {'cardinality': 337, 'dimension': 42}, 'value_count': {'min': 0, 'max': 20}}, 'dtype': DType(name='int64', element_type=<ElementType.Int: 'int'>, element_size=64, element_unit=None, signed=True, shape=Shape(dims=(Dimension(min=0, max=None), Dimension(min=0, max=20)))), 'is_list': True, 'is_ragged': True}, {'name': 'day_index', 'tags': {<Tags.CATEGORICAL: 'categorical'>}, 'properties': {}, 'dtype': DType(name='int64', element_type=<ElementType.Int: 'int'>, element_size=64, element_unit=None, signed=True, shape=Shape(dims=(Dimension(min=0, max=None),))), 'is_list': False, 'is_ragged': False}]"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "workflow.output_schema"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "70b05fd6",
   "metadata": {},
   "source": [
    "#### Save the preprocessing workflow"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "2d035a88-2146-4b9a-96fd-dd42be86e2a1",
   "metadata": {},
   "outputs": [],
   "source": [
    "workflow.save(os.path.join(DATA_FOLDER, \"workflow_etl\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "551018fc",
   "metadata": {},
   "source": [
    "### Export pre-processed data by day"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9515d625",
   "metadata": {},
   "source": [
    "In this example we are going to split the preprocessed parquet files by days, to allow for temporal training and evaluation. There will be a folder for each day and three parquet files within each day: `train.parquet`, `validation.parquet` and `test.parquet`.\n",
    "  \n",
    "P.s. It is worthwhile to note that the dataset has a single categorical feature (category), which, however, is inconsistent over time in the dataset. All interactions before day 84 (2014-06-23) have the same value for that feature, whereas many other categories are introduced afterwards. Thus for this example, we save only the last five days."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "2b4f5b73-459c-4356-87c8-9afb974cc77d",
   "metadata": {},
   "outputs": [],
   "source": [
    "# read in the processed train dataset\n",
    "sessions_gdf = cudf.read_parquet(os.path.join(DATA_FOLDER, \"processed_nvt/part_0.parquet\"))\n",
    "if USE_SYNTHETIC:\n",
    "    THRESHOLD_DAY_INDEX = int(os.environ.get(\"THRESHOLD_DAY_INDEX\", '1'))\n",
    "    sessions_gdf = sessions_gdf[sessions_gdf.day_index>=THRESHOLD_DAY_INDEX]\n",
    "else:\n",
    "    sessions_gdf = sessions_gdf[sessions_gdf.day_index>=178]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "e18d9c63",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "         session_id  item_id-count  \\\n",
      "6606147    11255549             12   \n",
      "6606148    11255552              2   \n",
      "6606149    11255553              2   \n",
      "\n",
      "                                              item_id-list  \\\n",
      "6606147  [605, 879, 743, 91, 4778, 1584, 3447, 8084, 34...   \n",
      "6606148                                       [185, 12289]   \n",
      "6606149                                       [7300, 1954]   \n",
      "\n",
      "                                     et_dayofweek_sin-list  \\\n",
      "6606147  [-0.43388454782514785, -0.43388454782514785, -...   \n",
      "6606148       [-0.43388454782514785, -0.43388454782514785]   \n",
      "6606149         [-0.7818309228245777, -0.7818309228245777]   \n",
      "\n",
      "                        product_recency_days_log_norm-list  \\\n",
      "6606147  [1.5241561, 1.523876, 1.523935, 1.5241641, 1.5...   \n",
      "6606148                              [-0.533007, 1.521495]   \n",
      "6606149                             [1.5338274, 1.5355083]   \n",
      "\n",
      "                                category-list  day_index  \n",
      "6606147  [4, 4, 4, 4, 4, 4, 4, 4, 4, 1, 4, 4]        178  \n",
      "6606148                                [1, 3]        178  \n",
      "6606149                                [8, 8]        180  \n"
     ]
    }
   ],
   "source": [
    "print(sessions_gdf.head(3))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "5175aeaf",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Creating time-based splits: 100%|██████████| 5/5 [00:02<00:00,  2.24it/s]\n"
     ]
    }
   ],
   "source": [
    "from transformers4rec.utils.data_utils import save_time_based_splits\n",
    "save_time_based_splits(data=nvt.Dataset(sessions_gdf),\n",
    "                       output_dir=os.path.join(DATA_FOLDER, \"preproc_sessions_by_day\"),\n",
    "                       partition_col='day_index',\n",
    "                       timestamp_col='session_id', \n",
    "                      )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "3bd1bad9",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "748"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# free gpu memory\n",
    "del  sessions_gdf\n",
    "gc.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "64772bf1",
   "metadata": {},
   "source": [
    "That's it! We created our sequential features, now we can go to the next notebook to train a PyTorch session-based model."
   ]
  }
 ],
 "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.8.10"
  },
  "vscode": {
   "interpreter": {
    "hash": "d795d7ca5d3ec3bd6293cc80853205a74ce23d484a2b8f537732a716747107c8"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}