examples/tutorial/02-ETL-with-NVTabular.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "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",
   "metadata": {},
   "source": [
    "<img src=\"https://developer.download.nvidia.com/notebooks/dlsw-notebooks/merlin_transformers4rec_tutorial-02-etl-with-nvtabular/nvidia_logo.png\" style=\"width: 90px; float: right;\">\n",
    "\n",
    "# ETL with NVTabular\n",
    "\n",
    "You can execute these tutorial notebooks 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",
    "\n",
    "This script will mount your local data folder that includes your data files to `/workspace/data` directory in the merlin-pytorch docker container.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1. Introduction"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In this notebook, we will create a preprocessing and feature engineering pipeline with [Rapids cuDF](https://github.com/rapidsai/cudf) and [Merlin NVTabular](https://github.com/NVIDIA-Merlin/NVTabular) libraries to prepare our dataset for session-based recommendation model training. \n",
    "\n",
    "NVTabular is a feature engineering and preprocessing library for tabular data that is designed to easily manipulate terabyte scale datasets and train deep learning (DL) based recommender systems. It provides high-level abstraction to simplify code and accelerates computation on the GPU using the RAPIDS Dask-cuDF library, and is designed to be interoperable with both PyTorch and TensorFlow using dataloaders that have been developed as extensions of native framework code.\n",
    "\n",
    "Our main goal is to create sequential features. In order to do that, we are going to perform the following:\n",
    "\n",
    "- Categorify categorical features with `Categorify()` op\n",
    "- Create temporal features with a `user-defined custom` op and `Lambda` op\n",
    "- Transform continuous features using `Log` and `Normalize` ops\n",
    "- Group all these features together at the session level sorting the interactions by time with `Groupby`\n",
    "- Finally export the preprocessed datasets to parquet files by hive-partitioning."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 1.1. Dataset"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In our hands-on exercise notebooks we are going to use a subset of the publicly available [eCommerce dataset](https://www.kaggle.com/mkechinov/ecommerce-behavior-data-from-multi-category-store). The eCommerce behavior data contains 7 months of data (from October 2019 to April 2020) from a large multi-category online store. Each row in the file represents an event. All events are related to products and users. Each event is like many-to-many relation between products and users.\n",
    "\n",
    "Data collected by Open CDP project and the source of the dataset is [REES46 Marketing Platform](https://rees46.com/)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2. Import Libraries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/usr/local/lib/python3.8/dist-packages/tqdm/auto.py:21: TqdmWarning: IProgress not found. Please update jupyter and ipywidgets. See https://ipywidgets.readthedocs.io/en/stable/user_install.html\n",
      "  from .autonotebook import tqdm as notebook_tqdm\n",
      "/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",
    "\n",
    "import numpy as np \n",
    "import cupy as cp\n",
    "import glob\n",
    "\n",
    "import cudf\n",
    "import nvtabular as nvt\n",
    "\n",
    "from nvtabular.ops import Operator\n",
    "from merlin.dag import ColumnSelector\n",
    "from merlin.schema import Schema, Tags"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "# avoid numba warnings\n",
    "from numba import config\n",
    "config.CUDA_LOW_OCCUPANCY_WARNINGS = 0"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 3. Set up Input and Output Data Paths"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "# define data path about where to get our data\n",
    "INPUT_DATA_DIR = os.environ.get(\"INPUT_DATA_DIR\", \"/workspace/data/\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 4. Read the Input Parquet file"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Even though the original dataset contains 7 months data files, we are going to use the first seven days of the `Oct-2019.csv` ecommerce dataset. We already performed certain preprocessing steps on the first month (Oct-2019) of the raw dataset in the `01-preprocess` notebook: <br>\n",
    "\n",
    "- we created `event_time_ts` column from `event_time` column which shows the time when event happened at (in UTC).\n",
    "- we created `prod_first_event_time_ts` column which indicates the timestamp that an item was seen first time.\n",
    "- we removed the rows where the `user_session` is Null. As a result, 2 rows were removed.\n",
    "- we categorified the `user_session` column, so that it now has only integer values.\n",
    "- we removed consequetively repeated (user, item) interactions. For example, an original session with `[1, 2, 4, 1, 2, 2, 3, 3, 3]` product interactions has become `[1, 2, 4, 1, 2, 3]` after removing the repeated interactions on the same item within the same session."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Below, we start by reading in `Oct-2019.parquet`with cuDF. In order to create and save `Oct-2019.parquet` file, please run [01-preprocess.ipynb](https://github.com/NVIDIA-Merlin/Transformers4Rec/blob/stable/examples/tutorial/01-preprocess.ipynb) notebook first."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "   user_session event_type  product_id          category_id category_code  \\\n",
      "0            43       view     5300797  2053013563173241677          <NA>   \n",
      "1            43       view     5300798  2053013563173241677          <NA>   \n",
      "2            43       view     5300284  2053013563173241677          <NA>   \n",
      "3            43       view     5300382  2053013563173241677          <NA>   \n",
      "4            43       view     5300366  2053013563173241677          <NA>   \n",
      "\n",
      "       brand  price    user_id  event_time_ts  prod_first_event_time_ts  \n",
      "0  panasonic  39.90  513903572     1570460611                1569948287  \n",
      "1  panasonic  32.18  513903572     1570460616                1569934097  \n",
      "2    rowenta  30.86  513903572     1570460621                1569927253  \n",
      "3  remington  28.22  513903572     1570460636                1570026747  \n",
      "4    polaris  26.46  513903572     1570460650                1570097085  \n",
      "CPU times: user 1.9 s, sys: 596 ms, total: 2.5 s\n",
      "Wall time: 2.52 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "df = cudf.read_parquet(os.path.join(INPUT_DATA_DIR, 'Oct-2019.parquet'))  \n",
    "print(df.head(5))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(6390928, 10)"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.shape"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's check if there is any column with nulls."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "user_session                False\n",
       "event_type                  False\n",
       "product_id                  False\n",
       "category_id                 False\n",
       "category_code                True\n",
       "brand                        True\n",
       "price                       False\n",
       "user_id                     False\n",
       "event_time_ts               False\n",
       "prod_first_event_time_ts    False\n",
       "dtype: bool"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.isnull().any()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "tags": []
   },
   "source": [
    "We see that `'category_code'` and `'brand'` columns have null values, and in the following cell we are going to fill these nulls with via categorify op, and then all categorical columns will be encoded to continuous integers. 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`."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 5. Initialize NVTabular Workflow\n",
    "\n",
    "### 5.1. Categorical Features Encoding"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# categorify features \n",
    "item_id = ['product_id'] >> nvt.ops.TagAsItemID()\n",
    "cat_feats = item_id + ['category_code', 'brand', 'user_id', 'category_id', 'event_type'] >> nvt.ops.Categorify()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 5.2. Extract Temporal Features"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# create time features\n",
    "session_ts = ['event_time_ts']\n",
    "\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",
    "\n",
    "sessiontime_weekday = (\n",
    "    session_time >> \n",
    "    nvt.ops.LambdaOp(lambda col: col.dt.weekday) >> \n",
    "    nvt.ops.Rename(name ='et_dayofweek')\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now let's create cycling features from the `sessiontime_weekday` column. We would like to use the temporal features (hour, day of week, month, etc.) that have inherently cyclical characteristic. We represent the day of week as a cycling feature (sine and cosine), so that it can be represented in a continuous space. That way, the difference between the representation of two different days is the same, in other words, with cyclical features we can convey closeness between data. You can read more about it [here](https://ianlondon.github.io/blog/encoding-cyclical-features-24hour-time/)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "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",
    "def get_cycled_feature_value_cos(col, max_value):\n",
    "    value_scaled = (col + 0.000001) / max_value\n",
    "    value_cos = np.cos(2*np.pi*value_scaled)\n",
    "    return value_cos"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "weekday_sin = (sessiontime_weekday >> \n",
    "               (lambda col: get_cycled_feature_value_sin(col+1, 7)) >> \n",
    "               nvt.ops.Rename(name = 'et_dayofweek_sin') >>\n",
    "               nvt.ops.AddMetadata(tags=[Tags.CONTINUOUS])\n",
    "              )\n",
    "    \n",
    "weekday_cos= (sessiontime_weekday >> \n",
    "              (lambda col: get_cycled_feature_value_cos(col+1, 7)) >> \n",
    "              nvt.ops.Rename(name = 'et_dayofweek_cos') >>\n",
    "              nvt.ops.AddMetadata(tags=[Tags.CONTINUOUS])\n",
    "             )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 5.2.1 Add Product Recency feature"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Let's define a custom op to calculate product recency in days"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 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['prod_first_event_time_ts']\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 [\"prod_first_event_time_ts\"]\n",
    "\n",
    "    @property\n",
    "    def output_dtype(self):\n",
    "        return np.float64"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "recency_features = ['event_time_ts'] >> ItemRecency() \n",
    "recency_features_norm = (recency_features >> \n",
    "                         nvt.ops.LogOp() >> \n",
    "                         nvt.ops.Normalize(out_dtype=np.float32) >> \n",
    "                         nvt.ops.Rename(name='product_recency_days_log_norm')\n",
    "                        )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "time_features = (\n",
    "    session_time +\n",
    "    sessiontime_weekday +\n",
    "    weekday_sin +\n",
    "    weekday_cos +\n",
    "    recency_features_norm\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 5.3. Normalize Continuous Features¶"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Smoothing price long-tailed distribution and applying standardization\n",
    "price_log = ['price'] >> nvt.ops.LogOp() >> nvt.ops.Normalize(out_dtype=np.float32) >> nvt.ops.Rename(name='price_log_norm')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Relative price to the average price for the category_id\n",
    "def relative_price_to_avg_categ(col, gdf):\n",
    "    epsilon = 1e-5\n",
    "    col = ((gdf['price'] - col) / (col + epsilon)) * (col > 0).astype(int)\n",
    "    return col\n",
    "    \n",
    "avg_category_id_pr = ['category_id'] >> nvt.ops.JoinGroupby(cont_cols =['price'], stats=[\"mean\"]) >> nvt.ops.Rename(name='avg_category_id_price')\n",
    "relative_price_to_avg_category = (\n",
    "    avg_category_id_pr >> \n",
    "    nvt.ops.LambdaOp(relative_price_to_avg_categ, dependency=['price']) >> \n",
    "    nvt.ops.Rename(name=\"relative_price_to_avg_categ_id\") >>\n",
    "    nvt.ops.AddMetadata(tags=[Tags.CONTINUOUS])\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 5.4. Grouping interactions into sessions"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Aggregate by session id and creates the sequential features"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "groupby_feats = ['event_time_ts', 'user_session'] + cat_feats + time_features + price_log + relative_price_to_avg_category"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Define Groupby Workflow\n",
    "groupby_features = groupby_feats >> nvt.ops.Groupby(\n",
    "    groupby_cols=[\"user_session\"], \n",
    "    sort_cols=[\"event_time_ts\"],\n",
    "    aggs={\n",
    "        'user_id': ['first'],\n",
    "        'product_id': [\"list\", \"count\"],\n",
    "        'category_code': [\"list\"],  \n",
    "        'brand': [\"list\"], \n",
    "        'category_id': [\"list\"], \n",
    "        'event_time_ts': [\"first\"],\n",
    "        'event_time_dt': [\"first\"],\n",
    "        'et_dayofweek_sin': [\"list\"],\n",
    "        'et_dayofweek_cos': [\"list\"],\n",
    "        'price_log_norm': [\"list\"],\n",
    "        'relative_price_to_avg_categ_id': [\"list\"],\n",
    "        'product_recency_days_log_norm': [\"list\"]\n",
    "        },\n",
    "    name_sep=\"-\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Select columns which are list"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "groupby_features_list = groupby_features['product_id-list',\n",
    "        'category_code-list',  \n",
    "        'brand-list', \n",
    "        'category_id-list', \n",
    "        'et_dayofweek_sin-list',\n",
    "        'et_dayofweek_cos-list',\n",
    "        'price_log_norm-list',\n",
    "        'relative_price_to_avg_categ_id-list',\n",
    "        'product_recency_days_log_norm-list']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "SESSIONS_MAX_LENGTH = 20 \n",
    "MINIMUM_SESSION_LENGTH = 2"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We truncate the sequence features in length according to sessions_max_length param, which is set as 20 in our example."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "groupby_features_trim = groupby_features_list >> nvt.ops.ListSlice(-SESSIONS_MAX_LENGTH, pad=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Create a `day_index` column in order to partition sessions by day when saving the parquet files."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [],
   "source": [
    "# calculate session day index based on 'timestamp-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",
    "            )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Select certain columns to be used in model training"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "sess_id = groupby_features['user_session'] >> nvt.ops.AddMetadata(tags=[Tags.CATEGORICAL])\n",
    "\n",
    "selected_features = sess_id + groupby_features['product_id-count'] + groupby_features_trim + day_index"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Filter out the session that have less than 2 interactions."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "filtered_sessions = selected_features >> nvt.ops.Filter(f=lambda df: df[\"product_id-count\"] >= MINIMUM_SESSION_LENGTH)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Initialize the NVTabular dataset object and workflow graph."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "NVTabular's preprocessing and feature engineering workflows are directed graphs of operators. When we initialize a Workflow with our pipeline, workflow organizes the input and output columns."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/usr/local/lib/python3.8/dist-packages/merlin/schema/tags.py:149: UserWarning: Compound tags like Tags.ITEM_ID have been deprecated and will be removed in a future version. Please use the atomic versions of these tags, like [<Tags.ITEM: 'item'>, <Tags.ID: 'id'>].\n",
      "  warnings.warn(\n",
      "/usr/local/lib/python3.8/dist-packages/merlin/schema/tags.py:149: UserWarning: Compound tags like Tags.ITEM_ID have been deprecated and will be removed in a future version. Please use the atomic versions of these tags, like [<Tags.ITEM: 'item'>, <Tags.ID: 'id'>].\n",
      "  warnings.warn(\n"
     ]
    }
   ],
   "source": [
    "workflow = nvt.Workflow(filtered_sessions)\n",
    "dataset = nvt.Dataset(df)\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(INPUT_DATA_DIR, \"processed_nvt\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "workflow.output_schema"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Above, we created an NVTabular Dataset object using our input dataset. Then, we calculate statistics for this workflow on the input dataset, i.e. on our training set, using the `workflow.fit()` method so that our Workflow can use these stats to transform any given input."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 6. Exporting data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We export dataset to parquet partitioned by the session `day_index` column."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [],
   "source": [
    "# define partition column\n",
    "PARTITION_COL = 'day_index'\n",
    "\n",
    "# define output_folder to store the partitioned parquet files\n",
    "OUTPUT_FOLDER = os.environ.get(\"OUTPUT_FOLDER\", INPUT_DATA_DIR + \"sessions_by_day\")\n",
    "!mkdir -p $OUTPUT_FOLDER"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In this section we are going to create a folder structure as shown below. As we explained above, this is just to structure parquet files so that it would be easier to do incremental training and evaluation."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```\n",
    "/sessions_by_day/\n",
    "|-- 1\n",
    "|   |-- train.parquet\n",
    "|   |-- valid.parquet\n",
    "|   |-- test.parquet\n",
    "\n",
    "|-- 2\n",
    "|   |-- train.parquet\n",
    "|   |-- valid.parquet\n",
    "|   |-- test.parquet\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "`gpu_preprocessing` function converts the process df to a Dataset object and write out hive-partitioned data to disk."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# read in the processed train dataset\n",
    "sessions_gdf = cudf.read_parquet(os.path.join(INPUT_DATA_DIR, \"processed_nvt/part_0.parquet\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "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": 29,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "   user_session  product_id-count  \\\n",
      "0             1               779   \n",
      "1             6               316   \n",
      "\n",
      "                                     product_id-list  \\\n",
      "0  [7171, 13290, 19757, 20401, 19757, 20401, 2122...   \n",
      "1  [4093, 3427, 20536, 27530, 11262, 18532, 33985...   \n",
      "\n",
      "                                  category_code-list  \\\n",
      "0  [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, ...   \n",
      "1  [17, 17, 1, 1, 1, 1, 1, 2, 2, 2, 2, 2, 2, 2, 2...   \n",
      "\n",
      "                                          brand-list  \\\n",
      "0  [34, 231, 239, 34, 239, 34, 120, 120, 249, 231...   \n",
      "1  [50, 1, 36, 1, 1, 1, 1, 3, 2, 2, 2, 2, 2, 2, 1...   \n",
      "\n",
      "                                    category_id-list  \\\n",
      "0  [3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, ...   \n",
      "1  [36, 36, 163, 163, 163, 163, 163, 2, 2, 2, 2, ...   \n",
      "\n",
      "                               et_dayofweek_sin-list  \\\n",
      "0  [0.7818321, 0.7818321, 0.7818321, 0.7818321, 0...   \n",
      "1  [0.43388295, 0.43388295, 0.43388295, 0.4338829...   \n",
      "\n",
      "                               et_dayofweek_cos-list  \\\n",
      "0  [0.6234891, 0.6234891, 0.6234891, 0.6234891, 0...   \n",
      "1  [-0.90096927, -0.90096927, -0.90096927, -0.900...   \n",
      "\n",
      "                                 price_log_norm-list  \\\n",
      "0  [-0.543168, -0.5673179, -0.578279, -0.62369585...   \n",
      "1  [0.23708352, 0.24015874, -0.80294055, -0.66287...   \n",
      "\n",
      "                 relative_price_to_avg_categ_id-list  \\\n",
      "0  [0.12176345592788261, 0.08783709744899952, 0.0...   \n",
      "1  [-0.23209574328358615, -0.22911214248041414, -...   \n",
      "\n",
      "                  product_recency_days_log_norm-list  day_index  \n",
      "0  [1.1603423, 1.1818986, 1.0499662, 1.0900966, 1...          1  \n",
      "1  [-0.7081338, -0.69109577, -0.73886937, -0.7074...          2  \n"
     ]
    }
   ],
   "source": [
    "print(sessions_gdf.head(2))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Creating time-based splits: 100%|███████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 7/7 [00:02<00:00,  2.75it/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= OUTPUT_FOLDER,\n",
    "                       partition_col=PARTITION_COL,\n",
    "                       timestamp_col='user_session', \n",
    "                      )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1  2  3  4  5  6  7\n"
     ]
    }
   ],
   "source": [
    "# check out the OUTPUT_FOLDER\n",
    "!ls $OUTPUT_FOLDER"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Save NVTabular workflow to load at the inference step."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can uncomment the cell below and execute it to see the output schema generated from nvtabular workflow."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "workflow_path = os.path.join(INPUT_DATA_DIR, 'workflow_etl')\n",
    "workflow.save(workflow_path)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 7. Wrap Up "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "That's it! We finished our first task. We reprocessed our dataset and created new features to train a session-based recommendation model. Please shut down the kernel before moving on to the next notebook."
   ]
  }
 ],
 "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": 4
}