examples/Building-and-deploying-multi-stage-RecSys/01-Building-Recommender-Systems-with-Merlin.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "8c3403a6",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Copyright 2023 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": "ad9b5cc0-2110-464e-9773-003ffe7d216c",
   "metadata": {},
   "source": [
    "<img src=\"https://developer.download.nvidia.com/notebooks/dlsw-notebooks/merlin_merlin_01-building-recommender-systems-with-merlin/nvidia_logo.png\" style=\"width: 90px; float: right;\"> \n",
    "\n",
    "## Building Intelligent Recommender Systems with Merlin\n",
    "\n",
    "This notebook is created using the latest stable [merlin-tensorflow](https://catalog.ngc.nvidia.com/orgs/nvidia/teams/merlin/containers/merlin-tensorflow/tags) container. "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f9657308-2e08-49b4-8924-eace75a4634c",
   "metadata": {},
   "source": [
    "### Overview\n",
    "\n",
    "Recommender Systems (RecSys) are the engine of the modern internet and the catalyst for human decisions. Building a recommendation system is challenging because it requires multiple stages (data preprocessing, offline training, item retrieval, filtering, ranking, ordering, etc.) to work together seamlessly and efficiently. The biggest challenges for new practitioners are the lack of understanding around what RecSys look like in the real world, and the gap between examples of simple models and a production-ready end-to-end recommender systems."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "405280b0-3d48-43b6-ab95-d29be7a43e9e",
   "metadata": {},
   "source": [
    "The figure below represents a four-stage recommender systems. This is a more complex process than only training a single model and deploying it, and it is much more realistic and closer to what's happening in the real-world recommender production systems."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "27220153",
   "metadata": {},
   "source": [
    "![fourstage](../images/fourstages.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b27ffed1-4b4b-4b6f-b933-31e9f6c1b4e1",
   "metadata": {},
   "source": [
    "In these series of notebooks, we are going to showcase how we can deploy a four-stage recommender systems using Merlin Systems library easily on [Triton Inference Server](https://github.com/triton-inference-server/server). Let's go over the concepts in the figure briefly. \n",
    "- **Retrieval:** This is the step to narrow down millions of items into thousands of candidates. We are going to train a Two-Tower item retrieval model to retrieve the relevant top-K candidate items.\n",
    "- **Filtering:** This step is to exclude the already interacted  or undesirable items from the candidate items set or to apply business logic rules. Although this is an important step, for this example we skip this step.\n",
    "- **Scoring:** This is also known as ranking. Here the retrieved and filtered candidate items are being scored. We are going to train a ranking model to be able to use at our scoring step. \n",
    "- **Ordering:** At this stage, we can order the final set of items that we want to recommend to the user. Here, we’re able to align the output of the model with business needs, constraints, or criteria.\n",
    "\n",
    "To learn more about the four-stage recommender systems, you can listen to Even Oldridge's [Moving Beyond Recommender Models talk](https://www.youtube.com/watch?v=5qjiY-kLwFY&list=PL65MqKWg6XcrdN4TJV0K1PdLhF_Uq-b43&index=7) at KDD'21 and read more [in this blog post](https://eugeneyan.com/writing/system-design-for-discovery/)."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e26f3194-9f17-4fa7-8baa-14333f2a122a",
   "metadata": {},
   "source": [
    "### Learning objectives\n",
    "- Understanding four stages of recommender systems\n",
    "- Training retrieval and ranking models with Merlin Models\n",
    "- Setting up feature store and approximate nearest neighbours (ANN) search libraries\n",
    "- Deploying trained models to Triton Inference Server with Merlin Systems"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "58d8bd1f-fa29-4d4b-a320-c76538f2302f",
   "metadata": {},
   "source": [
    "In addition to NVIDIA Merlin libraries and the Triton Inference Server client library, we use two external libraries in these series of examples:\n",
    "\n",
    "- [Feast](https://docs.feast.dev/): an end-to-end open source feature store library for machine learning\n",
    "- [Faiss](https://github.com/facebookresearch/faiss): a library for efficient similarity search and clustering of dense vectors\n",
    "\n",
    "You can find more information about `Feast feature store` and `Faiss` libraries in the next notebook."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "46b7f3bd",
   "metadata": {},
   "source": [
    "### Import required libraries and functions"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6c1586d8-e5a6-40c3-b6bb-61a3e62fa34c",
   "metadata": {},
   "source": [
    "**Compatibility:**\n",
    "\n",
    "This notebook is developed and tested using the latest `merlin-tensorflow` container from the NVIDIA NGC catalog. To find the tag for the most recently-released container, refer to the [Merlin TensorFlow](https://catalog.ngc.nvidia.com/orgs/nvidia/teams/merlin/containers/merlin-tensorflow) page."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "2cd8cc8d-5cc7-4a9f-91e5-3deec6f1fe74",
   "metadata": {},
   "outputs": [],
   "source": [
    "# for running this example on GPU, install the following libraries\n",
    "# %pip install \"feast==0.31\" faiss-gpu\n",
    "\n",
    "# for running this example on CPU, uncomment the following lines\n",
    "# %pip install tensorflow-cpu \"feast==0.31\" faiss-cpu\n",
    "# %pip uninstall cudf\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "08cdbfcc",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2023-06-29 19:49:32.836544: I tensorflow/core/platform/cpu_feature_guard.cc:194] This TensorFlow binary is optimized with oneAPI Deep Neural Network Library (oneDNN) to use the following CPU instructions in performance-critical operations:  SSE3 SSE4.1 SSE4.2 AVX\n",
      "To enable them in other operations, rebuild TensorFlow with the appropriate compiler flags.\n",
      "/usr/local/lib/python3.8/dist-packages/merlin/dtypes/mappings/torch.py:43: UserWarning: PyTorch dtype mappings did not load successfully due to an error: No module named 'torch'\n",
      "  warn(f\"PyTorch dtype mappings did not load successfully due to an error: {exc.msg}\")\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "WARNING:tensorflow:Please fix your imports. Module tensorflow.python.training.tracking.data_structures has been moved to tensorflow.python.trackable.data_structures. The old module will be deleted in version 2.11.\n",
      "[INFO]: sparse_operation_kit is imported\n",
      "WARNING:tensorflow:Please fix your imports. Module tensorflow.python.training.tracking.base has been moved to tensorflow.python.trackable.base. The old module will be deleted in version 2.11.\n",
      "[SOK INFO] Import /usr/local/lib/python3.8/dist-packages/merlin_sok-1.1.4-py3.8-linux-x86_64.egg/sparse_operation_kit/lib/libsok_experiment.so\n",
      "[SOK INFO] Import /usr/local/lib/python3.8/dist-packages/merlin_sok-1.1.4-py3.8-linux-x86_64.egg/sparse_operation_kit/lib/libsok_experiment.so\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2023-06-29 19:49:37.094972: I tensorflow/core/platform/cpu_feature_guard.cc:194] This TensorFlow binary is optimized with oneAPI Deep Neural Network Library (oneDNN) to use the following CPU instructions in performance-critical operations:  SSE3 SSE4.1 SSE4.2 AVX\n",
      "To enable them in other operations, rebuild TensorFlow with the appropriate compiler flags.\n",
      "2023-06-29 19:49:38.134481: W tensorflow/core/common_runtime/gpu/gpu_bfc_allocator.cc:42] Overriding orig_value setting because the TF_FORCE_GPU_ALLOW_GROWTH environment variable is set. Original config value was 0.\n",
      "2023-06-29 19:49:38.134526: I tensorflow/core/common_runtime/gpu/gpu_device.cc:1621] Created device /job:localhost/replica:0/task:0/device:GPU:0 with 24576 MB memory:  -> device: 0, name: Quadro RTX 8000, pci bus id: 0000:15:00.0, compute capability: 7.5\n",
      "2023-06-29 19:49:38.135533: W tensorflow/core/common_runtime/gpu/gpu_bfc_allocator.cc:42] Overriding orig_value setting because the TF_FORCE_GPU_ALLOW_GROWTH environment variable is set. Original config value was 0.\n",
      "2023-06-29 19:49:38.135562: I tensorflow/core/common_runtime/gpu/gpu_device.cc:1621] Created device /job:localhost/replica:0/task:0/device:GPU:1 with 24576 MB memory:  -> device: 1, name: Quadro RTX 8000, pci bus id: 0000:2d:00.0, compute capability: 7.5\n",
      "/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"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[SOK INFO] Initialize finished, communication tool: horovod\n"
     ]
    }
   ],
   "source": [
    "import os\n",
    "import nvtabular as nvt\n",
    "from nvtabular.ops import Rename, Filter, Dropna, LambdaOp, Categorify, \\\n",
    "    TagAsUserFeatures, TagAsUserID, TagAsItemFeatures, TagAsItemID, AddMetadata\n",
    "\n",
    "from merlin.schema.tags import Tags\n",
    "from merlin.dag.ops.subgraph import Subgraph\n",
    "import merlin.models.tf as mm\n",
    "from merlin.io.dataset import Dataset\n",
    "from merlin.datasets.ecommerce import transform_aliccp\n",
    "import tensorflow as tf\n",
    "\n",
    "# for running this example on CPU, comment out the line below\n",
    "os.environ[\"TF_GPU_ALLOCATOR\"] = \"cuda_malloc_async\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "028a1398-76a8-4998-97d8-34a806e130d3",
   "metadata": {},
   "outputs": [],
   "source": [
    "# disable INFO and DEBUG logging everywhere\n",
    "import logging\n",
    "\n",
    "logging.disable(logging.WARNING)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "baad8ae3",
   "metadata": {},
   "source": [
    "In this example notebook, we will generate the synthetic train and test datasets mimicking the real [Ali-CCP: Alibaba Click and Conversion Prediction](https://tianchi.aliyun.com/dataset/dataDetail?dataId=408#1) dataset to build our recommender system models.\n",
    "\n",
    "First, we define our input path and feature repo path."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "81ddb370",
   "metadata": {},
   "outputs": [],
   "source": [
    "DATA_FOLDER = os.environ.get(\"DATA_FOLDER\", \"/workspace/data/\")\n",
    "# set up the base dir for feature store\n",
    "BASE_DIR = os.environ.get(\n",
    "    \"BASE_DIR\", \"/Merlin/examples/Building-and-deploying-multi-stage-RecSys/\"\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7a746a3f-1845-4af3-8a37-1b34aa1bb81b",
   "metadata": {},
   "source": [
    "Then, we use `generate_data` utility function to generate synthetic dataset. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "b44b3378-7297-4946-a271-742a9239bc3e",
   "metadata": {},
   "outputs": [],
   "source": [
    "from merlin.datasets.synthetic import generate_data\n",
    "\n",
    "NUM_ROWS = os.environ.get(\"NUM_ROWS\", 100_000)\n",
    "train_raw, valid_raw = generate_data(\"aliccp-raw\", int(NUM_ROWS), set_sizes=(0.7, 0.3))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "16bae6de-0345-4963-8f73-3ae234e54040",
   "metadata": {},
   "source": [
    "If you would like to use the real ALI-CCP dataset, you can use [get_aliccp()](https://github.com/NVIDIA-Merlin/models/blob/stable/merlin/datasets/ecommerce/aliccp/dataset.py) function instead. This function takes the raw csv files, and generate parquet files that can be directly fed to NVTabular workflow above."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7bd843be-dfba-4f8b-bac1-608e6571352d",
   "metadata": {},
   "source": [
    "### Set up a feature store with Feast"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c543b71c-6ba2-4e43-8779-8bffb62d2cee",
   "metadata": {},
   "source": [
    "Before we move onto the next step, we need to create a Feast feature repository. [Feast](https://feast.dev/) is an end-to-end open source feature store for machine learning. Feast (Feature Store) is a customizable operational data system that re-uses existing infrastructure to manage and serve machine learning features to real-time models.\n",
    "\n",
    "We will create the feature repo in the current working directory, which is `BASE_DIR` for us."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "2e7e96d2-9cd2-40d1-b356-8cd76b57bb4a",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Creating a new Feast repository in \u001b[1m\u001b[32m/raid/workshared/merlin/examples/Building-and-deploying-multi-stage-RecSys/feast_repo\u001b[0m.\n",
      "\n"
     ]
    }
   ],
   "source": [
    "!rm -rf $BASE_DIR/feast_repo\n",
    "!cd $BASE_DIR && feast init feast_repo"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f6d4d773-144e-4e34-82cd-f2b50fce601c",
   "metadata": {},
   "source": [
    "You should be seeing a message like <i>Creating a new Feast repository in ... </i> printed out above. Now, navigate to the `feature_repo` folder and remove the demo parquet file created by default, and `examples.py` file."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "26ba2521-ed1b-4c2b-afdd-26b4a5a9c008",
   "metadata": {},
   "outputs": [],
   "source": [
    "feature_repo_path = os.path.join(BASE_DIR, \"feast_repo/feature_repo\")\n",
    "if os.path.exists(f\"{feature_repo_path}/example_repo.py\"):\n",
    "    os.remove(f\"{feature_repo_path}/example_repo.py\")\n",
    "if os.path.exists(f\"{feature_repo_path}/data/driver_stats.parquet\"):\n",
    "    os.remove(f\"{feature_repo_path}/data/driver_stats.parquet\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "24ae0e29-c156-4df9-8977-238786160a8c",
   "metadata": {},
   "source": [
    "### Exporting user and item features"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "ea0b369c-2f01-42e3-9f3c-74c3ff4a6d64",
   "metadata": {},
   "outputs": [],
   "source": [
    "from merlin.models.utils.dataset import unique_rows_by_features\n",
    "\n",
    "user_features = (\n",
    "    unique_rows_by_features(train_raw, Tags.USER, Tags.USER_ID)\n",
    "    .compute()\n",
    "    .reset_index(drop=True)\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4f2d12f5-c753-4392-b113-965d97d2fe35",
   "metadata": {},
   "source": [
    "We will artificially add `datetime` and `created` timestamp columns to our user_features dataframe. This required by Feast to track the user-item features and their creation time and to determine which version to use when we query Feast."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "d30bd2f8-8a78-4df7-9bc4-42bd741c5b99",
   "metadata": {},
   "outputs": [],
   "source": [
    "from datetime import datetime\n",
    "\n",
    "user_features[\"datetime\"] = datetime.now()\n",
    "user_features[\"datetime\"] = user_features[\"datetime\"].astype(\"datetime64[ns]\")\n",
    "user_features[\"created\"] = datetime.now()\n",
    "user_features[\"created\"] = user_features[\"created\"].astype(\"datetime64[ns]\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "d4998cd1-9dcd-4911-8f23-372e197b41e9",
   "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>user_id</th>\n",
       "      <th>user_shops</th>\n",
       "      <th>user_profile</th>\n",
       "      <th>user_group</th>\n",
       "      <th>user_gender</th>\n",
       "      <th>user_age</th>\n",
       "      <th>user_consumption_1</th>\n",
       "      <th>user_consumption_2</th>\n",
       "      <th>user_is_occupied</th>\n",
       "      <th>user_geography</th>\n",
       "      <th>user_intentions</th>\n",
       "      <th>user_brands</th>\n",
       "      <th>user_categories</th>\n",
       "      <th>datetime</th>\n",
       "      <th>created</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>6</th>\n",
       "      <td>7</td>\n",
       "      <td>590</td>\n",
       "      <td>1</td>\n",
       "      <td>1</td>\n",
       "      <td>1</td>\n",
       "      <td>1</td>\n",
       "      <td>1</td>\n",
       "      <td>1</td>\n",
       "      <td>1</td>\n",
       "      <td>1</td>\n",
       "      <td>171</td>\n",
       "      <td>293</td>\n",
       "      <td>31</td>\n",
       "      <td>2023-06-29 19:49:50.300270</td>\n",
       "      <td>2023-06-29 19:49:50.303330</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   user_id  user_shops  user_profile  user_group  user_gender  user_age  \\\n",
       "6        7         590             1           1            1         1   \n",
       "\n",
       "   user_consumption_1  user_consumption_2  user_is_occupied  user_geography  \\\n",
       "6                   1                   1                 1               1   \n",
       "\n",
       "   user_intentions  user_brands  user_categories                   datetime  \\\n",
       "6              171          293               31 2023-06-29 19:49:50.300270   \n",
       "\n",
       "                     created  \n",
       "6 2023-06-29 19:49:50.303330  "
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "user_features[user_features[\"user_id\"] == 7]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "2981b3ed-6156-49f0-aa14-326a3853a58a",
   "metadata": {},
   "outputs": [],
   "source": [
    "user_features.to_parquet(\n",
    "    os.path.join(feature_repo_path, \"data\", \"user_features.parquet\")\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "0a33a668-8e2a-4546-8f54-0060d405ba91",
   "metadata": {},
   "outputs": [],
   "source": [
    "item_features = (\n",
    "    unique_rows_by_features(train_raw, Tags.ITEM, Tags.ITEM_ID)\n",
    "    .compute()\n",
    "    .reset_index(drop=True)\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "68a694d6-926f-4b0f-8edc-8cc7ac85ade7",
   "metadata": {},
   "outputs": [],
   "source": [
    "item_features[\"datetime\"] = datetime.now()\n",
    "item_features[\"datetime\"] = item_features[\"datetime\"].astype(\"datetime64[ns]\")\n",
    "item_features[\"created\"] = datetime.now()\n",
    "item_features[\"created\"] = item_features[\"created\"].astype(\"datetime64[ns]\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "6c03fa22-b112-4243-bbe1-1cd7260cb85b",
   "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>item_id</th>\n",
       "      <th>item_category</th>\n",
       "      <th>item_shop</th>\n",
       "      <th>item_brand</th>\n",
       "      <th>item_intention</th>\n",
       "      <th>datetime</th>\n",
       "      <th>created</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>1</td>\n",
       "      <td>1</td>\n",
       "      <td>1</td>\n",
       "      <td>1</td>\n",
       "      <td>1</td>\n",
       "      <td>2023-06-29 19:49:50.410715</td>\n",
       "      <td>2023-06-29 19:49:50.412307</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>2</td>\n",
       "      <td>6</td>\n",
       "      <td>412</td>\n",
       "      <td>142</td>\n",
       "      <td>66</td>\n",
       "      <td>2023-06-29 19:49:50.410715</td>\n",
       "      <td>2023-06-29 19:49:50.412307</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>3</td>\n",
       "      <td>12</td>\n",
       "      <td>824</td>\n",
       "      <td>284</td>\n",
       "      <td>132</td>\n",
       "      <td>2023-06-29 19:49:50.410715</td>\n",
       "      <td>2023-06-29 19:49:50.412307</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>4</td>\n",
       "      <td>18</td>\n",
       "      <td>1236</td>\n",
       "      <td>426</td>\n",
       "      <td>197</td>\n",
       "      <td>2023-06-29 19:49:50.410715</td>\n",
       "      <td>2023-06-29 19:49:50.412307</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>5</td>\n",
       "      <td>24</td>\n",
       "      <td>1648</td>\n",
       "      <td>568</td>\n",
       "      <td>263</td>\n",
       "      <td>2023-06-29 19:49:50.410715</td>\n",
       "      <td>2023-06-29 19:49:50.412307</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   item_id  item_category  item_shop  item_brand  item_intention  \\\n",
       "0        1              1          1           1               1   \n",
       "1        2              6        412         142              66   \n",
       "2        3             12        824         284             132   \n",
       "3        4             18       1236         426             197   \n",
       "4        5             24       1648         568             263   \n",
       "\n",
       "                    datetime                    created  \n",
       "0 2023-06-29 19:49:50.410715 2023-06-29 19:49:50.412307  \n",
       "1 2023-06-29 19:49:50.410715 2023-06-29 19:49:50.412307  \n",
       "2 2023-06-29 19:49:50.410715 2023-06-29 19:49:50.412307  \n",
       "3 2023-06-29 19:49:50.410715 2023-06-29 19:49:50.412307  \n",
       "4 2023-06-29 19:49:50.410715 2023-06-29 19:49:50.412307  "
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "item_features.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "c312884b-a1f8-4e08-8068-696e06a9bf46",
   "metadata": {},
   "outputs": [],
   "source": [
    "# save to disk\n",
    "item_features.to_parquet(\n",
    "    os.path.join(feature_repo_path, \"data\", \"item_features.parquet\")\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2e428d01-f2f0-42d4-85d0-0986bb83a847",
   "metadata": {},
   "source": [
    "### Feature Engineering with NVTabular"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "d4bf870c-30cf-4074-88d3-b75981b3a873",
   "metadata": {},
   "outputs": [],
   "source": [
    "output_path = os.path.join(DATA_FOLDER, \"processed_nvt\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1e7bfb5c-88ed-4cf9-8a17-98c0284adb36",
   "metadata": {},
   "source": [
    "In the following NVTabular workflow, notice that we apply the `Dropna()` Operator at the end. We add the Operator to remove rows with missing values in the final DataFrame after the preceding transformations. Although, the synthetic dataset that we generate and use in this notebook does not have null entries, you might have null entries in your `user_id` and `item_id` columns in your own custom dataset. Therefore, while applying `Dropna()` we will not be registering null `user_id_raw` and `item_id_raw` values in the feature store, and will be avoiding potential issues that can occur because of any null entries."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "f91ada78-4e4d-4415-ab94-e351aa454e9e",
   "metadata": {},
   "outputs": [],
   "source": [
    "user_id_raw = [\"user_id\"] >> Rename(postfix='_raw') >> LambdaOp(lambda col: col.astype(\"int32\")) >> TagAsUserFeatures()\n",
    "item_id_raw = [\"item_id\"] >> Rename(postfix='_raw') >> LambdaOp(lambda col: col.astype(\"int32\")) >> TagAsItemFeatures()\n",
    "\n",
    "\n",
    "item_cat = Categorify(dtype=\"int32\")\n",
    "items = ([\"item_id\",\"item_category\", \"item_shop\", \"item_brand\"] >> item_cat)\n",
    "\n",
    "subgraph_item = Subgraph(\n",
    "     \"item\", \n",
    "     Subgraph(\"items_cat\", items) + \n",
    "    (items[\"item_id\"] >> TagAsItemID()) + \n",
    "    (items[\"item_category\", \"item_shop\", \"item_brand\"] >> TagAsItemFeatures())\n",
    ")\n",
    "subgraph_user = Subgraph(\n",
    "    \"user\",\n",
    "    ([\"user_id\"] >> Categorify(dtype=\"int32\") >> TagAsUserID()) +\n",
    "    (\n",
    "        [\n",
    "            \"user_shops\",\n",
    "            \"user_profile\",\n",
    "            \"user_group\",\n",
    "            \"user_gender\",\n",
    "            \"user_age\",\n",
    "            \"user_consumption_2\",\n",
    "            \"user_is_occupied\",\n",
    "            \"user_geography\",\n",
    "            \"user_intentions\",\n",
    "            \"user_brands\",\n",
    "            \"user_categories\",\n",
    "        ] >> Categorify(dtype=\"int32\") >> TagAsUserFeatures()\n",
    "    )\n",
    ")\n",
    "\n",
    "targets = [\"click\"] >> AddMetadata(tags=[Tags.BINARY_CLASSIFICATION, \"target\"])\n",
    "outputs = subgraph_user + subgraph_item + targets\n",
    "\n",
    "# add dropna op to filter rows with nulls\n",
    "outputs = outputs >> Dropna()\n",
    "nvt_wkflow = nvt.Workflow(outputs)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "71aae006-a161-4127-889a-8f433a9f7362",
   "metadata": {},
   "source": [
    "Let's call `transform_aliccp` utility function to be able to perform `fit` and `transform` steps on the raw dataset applying the operators defined in the NVTabular workflow pipeline below, and also save our workflow model. After fit and transform, the processed parquet files are saved to output_path."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "814e8438-642a-4f03-baaf-44dab8d1b5e5",
   "metadata": {},
   "outputs": [],
   "source": [
    "transform_aliccp(\n",
    "    (train_raw, valid_raw), output_path, nvt_workflow=nvt_wkflow, workflow_name=\"workflow\"\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "09c87748-af61-42b8-8574-1afe3d71118f",
   "metadata": {},
   "source": [
    "### Training a Retrieval Model with Two-Tower Model"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e644fcba-7b0b-44c0-97fd-80f4fcb01191",
   "metadata": {},
   "source": [
    "We start with the offline candidate retrieval stage. We are going to train a Two-Tower model for item retrieval. To learn more about the Two-tower model you can visit [05-Retrieval-Model.ipynb](https://github.com/NVIDIA-Merlin/models/blob/stable/examples/05-Retrieval-Model.ipynb)."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cf9bca46-a6b6-4a73-afd8-fe2869c60748",
   "metadata": {},
   "source": [
    "#### Feature Engineering with NVTabular"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "da2b09cc-09fb-4814-a1cb-7e6168d9eb4b",
   "metadata": {},
   "source": [
    "We are going to process our raw categorical features by encoding them using `Categorify()` operator and tag the features with `user` or `item` tags in the schema file. To learn more about [NVTabular](https://github.com/NVIDIA-Merlin/NVTabular) and the schema object visit this example [notebook](https://github.com/NVIDIA-Merlin/models/blob/stable/examples/02-Merlin-Models-and-NVTabular-integration.ipynb) in the Merlin Models repo."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f3bc7abd-8d97-452b-a4af-5227821a99c9",
   "metadata": {},
   "source": [
    "Define a new output path to store the filtered datasets and schema files."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "df72a793-194b-44f4-80c3-aaa368a9a01e",
   "metadata": {},
   "outputs": [],
   "source": [
    "output_path2 = os.path.join(DATA_FOLDER, \"processed/retrieval\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "251d4697-8f9c-4c93-8de4-c3480a8378de",
   "metadata": {},
   "outputs": [],
   "source": [
    "train_tt = Dataset(os.path.join(output_path, \"train\", \"*.parquet\"))\n",
    "valid_tt = Dataset(os.path.join(output_path, \"valid\", \"*.parquet\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ffd7e2ac-a251-49d0-943b-e9272c852ba6",
   "metadata": {},
   "source": [
    "We select only positive interaction rows where `click==1` in the dataset with `Filter()` operator."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "7e085a6d-74ad-4c24-8e7c-4e449c15f471",
   "metadata": {},
   "outputs": [],
   "source": [
    "inputs = train_tt.schema.column_names\n",
    "outputs = inputs >> Filter(f=lambda df: df[\"click\"] == 1)\n",
    "\n",
    "nvt_wkflow.fit(train_tt)\n",
    "\n",
    "nvt_wkflow.transform(train_tt).to_parquet(\n",
    "    output_path=os.path.join(output_path2, \"train\")\n",
    ")\n",
    "\n",
    "nvt_wkflow.transform(valid_tt).to_parquet(\n",
    "    output_path=os.path.join(output_path2, \"valid\")\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cc4721ae-7228-4d3f-9586-dcdfefecc19f",
   "metadata": {},
   "source": [
    "NVTabular exported the schema file, `schema.pbtxt` a protobuf text file, of our processed dataset. To learn more about the schema object and schema file you can explore [02-Merlin-Models-and-NVTabular-integration.ipynb](https://github.com/NVIDIA-Merlin/models/blob/stable/examples/02-Merlin-Models-and-NVTabular-integration.ipynb) notebook."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "aa025b80-0f18-437c-a85f-4edcb89f4222",
   "metadata": {},
   "source": [
    "**Read filtered parquet files as Dataset objects.**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "252a8e60-b447-46b5-ade6-3557cbafa797",
   "metadata": {},
   "outputs": [],
   "source": [
    "train_tt = Dataset(os.path.join(output_path2, \"train\", \"*.parquet\"), part_size=\"500MB\")\n",
    "valid_tt = Dataset(os.path.join(output_path2, \"valid\", \"*.parquet\"), part_size=\"500MB\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "id": "71063653-2f39-4b54-8399-145d6f281d4d",
   "metadata": {},
   "outputs": [],
   "source": [
    "schema = train_tt.schema.select_by_tag([Tags.ITEM_ID, Tags.USER_ID, Tags.ITEM, Tags.USER]).without(['click'])\n",
    "train_tt.schema = schema\n",
    "valid_tt.schema = schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "id": "9312511a-f368-42f2-93d2-eb95aebbf46c",
   "metadata": {},
   "outputs": [],
   "source": [
    "model_tt = mm.TwoTowerModel(\n",
    "    schema,\n",
    "    query_tower=mm.MLPBlock([128, 64], no_activation_last_layer=True),\n",
    "    samplers=[mm.InBatchSampler()],\n",
    "    embedding_options=mm.EmbeddingOptions(infer_embedding_sizes=True),\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "id": "4d47cb8b-e06a-4932-9a19-fb244ef43152",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/usr/local/lib/python3.8/dist-packages/keras/initializers/initializers_v2.py:120: UserWarning: The initializer TruncatedNormal is unseeded and being called multiple times, which will return identical values  each time (even if the initializer is unseeded). Please update your code to provide a seed to the initializer, or avoid using the same initalizer instance more than once.\n",
      "  warnings.warn(\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "9/9 [==============================] - 11s 275ms/step - loss: 8.9538 - recall_at_10: 0.0101 - ndcg_at_10: 0.0067 - regularization_loss: 0.0000e+00 - loss_batch: 8.8711 - val_loss: 8.9179 - val_recall_at_10: 0.0212 - val_ndcg_at_10: 0.0155 - val_regularization_loss: 0.0000e+00 - val_loss_batch: 8.5806\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "<keras.callbacks.History at 0x7fd4b04139d0>"
      ]
     },
     "execution_count": 26,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "model_tt.compile(\n",
    "    optimizer=\"adam\",\n",
    "    run_eagerly=False,\n",
    "    loss=\"categorical_crossentropy\",\n",
    "    metrics=[mm.RecallAt(10), mm.NDCGAt(10)],\n",
    ")\n",
    "model_tt.fit(train_tt, validation_data=valid_tt, batch_size=1024 * 8, epochs=1)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "80d83007-f9e8-408f-9f65-a0e9e19cb586",
   "metadata": {},
   "source": [
    "### Exporting query (user) model"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "22af58a9-5525-454a-bf25-a9df0462aa53",
   "metadata": {},
   "source": [
    "We export the query tower to use it later during the model deployment stage with Merlin Systems."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "id": "d2370f13-ff9a-4ee0-ba1e-451c7bec0f8a",
   "metadata": {},
   "outputs": [],
   "source": [
    "query_tower = model_tt.retrieval_block.query_block()\n",
    "query_tower.save(os.path.join(BASE_DIR, \"query_tower\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e16401d4",
   "metadata": {
    "tags": []
   },
   "source": [
    "### Training a Ranking Model with DLRM"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b72e8a2a-fc4a-43ab-934c-6d941c56aad2",
   "metadata": {},
   "source": [
    "Now we will move onto training an offline ranking model. This ranking model will be used for scoring our retrieved items."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c4f2b234",
   "metadata": {},
   "source": [
    "Read processed parquet files. We use the `schema` object to define our model."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "id": "cb870461-6ac2-49b2-ba6a-2da6ecb57f1d",
   "metadata": {},
   "outputs": [],
   "source": [
    "# define train and valid dataset objects\n",
    "train = Dataset(os.path.join(output_path, \"train\", \"*.parquet\"), part_size=\"500MB\")\n",
    "valid = Dataset(os.path.join(output_path, \"valid\", \"*.parquet\"), part_size=\"500MB\")\n",
    "\n",
    "# define schema object\n",
    "schema = train.schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "id": "30e4ebc2",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'click'"
      ]
     },
     "execution_count": 29,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "target_column = schema.select_by_tag(Tags.TARGET).column_names[0]\n",
    "target_column"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8f68e26b",
   "metadata": {},
   "source": [
    "Deep Learning Recommendation Model [(DLRM)](https://arxiv.org/abs/1906.00091) architecture is a popular neural network model originally proposed by Facebook in 2019. The model was introduced as a personalization deep learning model that uses embeddings to process sparse features that represent categorical data and a multilayer perceptron (MLP) to process dense features, then interacts these features explicitly using the statistical techniques proposed in [here](https://ieeexplore.ieee.org/stamp/stamp.jsp?tp=&arnumber=5694074). To learn more about DLRM architetcture please visit `Exploring-different-models` [notebook](https://github.com/NVIDIA-Merlin/models/blob/stable/examples/04-Exporting-ranking-models.ipynb) in the Merlin Models GH repo."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "id": "e4325080",
   "metadata": {},
   "outputs": [],
   "source": [
    "model = mm.DLRMModel(\n",
    "    schema,\n",
    "    embedding_dim=64,\n",
    "    bottom_block=mm.MLPBlock([128, 64]),\n",
    "    top_block=mm.MLPBlock([128, 64, 32]),\n",
    "    prediction_tasks=mm.BinaryClassificationTask(target_column),\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "id": "bfe2aa9e",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "5/5 [==============================] - 5s 305ms/step - loss: 0.6932 - auc: 0.5005 - regularization_loss: 0.0000e+00 - loss_batch: 0.6932 - val_loss: 0.6931 - val_auc: 0.5029 - val_regularization_loss: 0.0000e+00 - val_loss_batch: 0.6931\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "<keras.callbacks.History at 0x7fd449398a30>"
      ]
     },
     "execution_count": 31,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "model.compile(optimizer=\"adam\", run_eagerly=False, metrics=[tf.keras.metrics.AUC()])\n",
    "model.fit(train, validation_data=valid, batch_size=16 * 1024)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "498c4d49-7a59-4260-87b9-b86b66f2c67f",
   "metadata": {},
   "source": [
    "Let's save our DLRM model to be able to load back at the deployment stage. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "id": "00447c12-ea80-4d98-ab47-cc1a982a6958",
   "metadata": {},
   "outputs": [],
   "source": [
    "model.save(os.path.join(BASE_DIR, \"dlrm\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d64a3f3f-81d8-489c-835f-c62f76df22d5",
   "metadata": {},
   "source": [
    "In the following cells we are going to export the required user and item features files, and save the query (user) tower model and item embeddings to disk. If you want to read more about exporting retrieval models, please visit [05-Retrieval-Model.ipynb](https://github.com/NVIDIA-Merlin/models/blob/stable/examples/05-Retrieval-Model.ipynb) notebook in Merlin Models library repo."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ff30ceab-b264-4509-9c5b-5a10425e143b",
   "metadata": {},
   "source": [
    "### Extract and save Item embeddings"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "id": "e62f65f8-e8f1-447e-9500-5960807c36f2",
   "metadata": {},
   "outputs": [],
   "source": [
    "from merlin.systems.dag.ops.tensorflow import PredictTensorflow\n",
    "from merlin.systems.dag.ops.workflow import TransformWorkflow\n",
    "\n",
    "workflow =  nvt.Workflow([\"item_id\"] + (['item_id', 'item_brand', 'item_category', 'item_shop'] >> TransformWorkflow(nvt_wkflow.get_subworkflow(\"item\")) >> PredictTensorflow(model_tt.first.item_block())))\n",
    "item_embeddings = workflow.fit_transform(Dataset(item_features)).to_ddf().compute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "id": "e02f0957-6665-400a-80c0-60b307466caf",
   "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>item_id</th>\n",
       "      <th>output_1</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>453</th>\n",
       "      <td>945</td>\n",
       "      <td>[0.012117806822061539, -0.02241620607674122, 0...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>454</th>\n",
       "      <td>948</td>\n",
       "      <td>[0.012117806822061539, -0.02241620607674122, 0...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>455</th>\n",
       "      <td>956</td>\n",
       "      <td>[0.012117806822061539, -0.02241620607674122, 0...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>456</th>\n",
       "      <td>1437</td>\n",
       "      <td>[0.012117806822061539, -0.02241620607674122, 0...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>457</th>\n",
       "      <td>1469</td>\n",
       "      <td>[0.012117806822061539, -0.02241620607674122, 0...</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "     item_id                                           output_1\n",
       "453      945  [0.012117806822061539, -0.02241620607674122, 0...\n",
       "454      948  [0.012117806822061539, -0.02241620607674122, 0...\n",
       "455      956  [0.012117806822061539, -0.02241620607674122, 0...\n",
       "456     1437  [0.012117806822061539, -0.02241620607674122, 0...\n",
       "457     1469  [0.012117806822061539, -0.02241620607674122, 0..."
      ]
     },
     "execution_count": 34,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "item_embeddings.tail()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "id": "66d7271e-0ea6-4568-ac5a-04089735f542",
   "metadata": {},
   "outputs": [],
   "source": [
    "# save to disk\n",
    "item_embeddings.to_parquet(os.path.join(BASE_DIR, \"item_embeddings.parquet\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "dadae279-913c-487b-ad55-4b4d6c110dc1",
   "metadata": {},
   "source": [
    "### Create feature definitions "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1f70939f-8063-4422-b29b-6668acb1cfb7",
   "metadata": {},
   "source": [
    "Now we will create our user and item features definitions in the user_features.py and item_features.py files and save these files in the feature_repo."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "id": "4ee27d67-e35a-42c5-8025-ed73f35c8e13",
   "metadata": {},
   "outputs": [],
   "source": [
    "file = open(os.path.join(feature_repo_path, \"user_features.py\"), \"w\")\n",
    "file.write(\n",
    "    \"\"\"\n",
    "from datetime import timedelta\n",
    "from feast import Entity, Field, FeatureView, ValueType\n",
    "from feast.types import Int32\n",
    "from feast.infra.offline_stores.file_source import FileSource\n",
    "\n",
    "user_features = FileSource(\n",
    "    path=\"{}\",\n",
    "    timestamp_field=\"datetime\",\n",
    "    created_timestamp_column=\"created\",\n",
    ")\n",
    "\n",
    "user = Entity(name=\"user_id\", value_type=ValueType.INT32, join_keys=[\"user_id\"],)\n",
    "\n",
    "user_features_view = FeatureView(\n",
    "    name=\"user_features\",\n",
    "    entities=[user],\n",
    "    ttl=timedelta(0),\n",
    "    schema=[\n",
    "        Field(name=\"user_shops\", dtype=Int32),\n",
    "        Field(name=\"user_profile\", dtype=Int32),\n",
    "        Field(name=\"user_group\", dtype=Int32),\n",
    "        Field(name=\"user_gender\", dtype=Int32),\n",
    "        Field(name=\"user_age\", dtype=Int32),\n",
    "        Field(name=\"user_consumption_2\", dtype=Int32),\n",
    "        Field(name=\"user_is_occupied\", dtype=Int32),\n",
    "        Field(name=\"user_geography\", dtype=Int32),\n",
    "        Field(name=\"user_intentions\", dtype=Int32),\n",
    "        Field(name=\"user_brands\", dtype=Int32),\n",
    "        Field(name=\"user_categories\", dtype=Int32),\n",
    "    ],\n",
    "    online=True,\n",
    "    source=user_features,\n",
    "    tags=dict(),\n",
    ")\n",
    "\"\"\".format(\n",
    "        os.path.join(feature_repo_path, \"data/\", \"user_features.parquet\")\n",
    "    )\n",
    ")\n",
    "file.close()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "id": "48a5927c-840d-410c-8f5b-bebce4f79640",
   "metadata": {},
   "outputs": [],
   "source": [
    "with open(os.path.join(feature_repo_path, \"item_features.py\"), \"w\") as f:\n",
    "    f.write(\n",
    "        \"\"\"\n",
    "from datetime import timedelta\n",
    "from feast import Entity, Field, FeatureView, ValueType\n",
    "from feast.types import Int32\n",
    "from feast.infra.offline_stores.file_source import FileSource\n",
    "\n",
    "item_features = FileSource(\n",
    "    path=\"{}\",\n",
    "    timestamp_field=\"datetime\",\n",
    "    created_timestamp_column=\"created\",\n",
    ")\n",
    "\n",
    "item = Entity(name=\"item_id\", value_type=ValueType.INT32, join_keys=[\"item_id\"],)\n",
    "\n",
    "item_features_view = FeatureView(\n",
    "    name=\"item_features\",\n",
    "    entities=[item],\n",
    "    ttl=timedelta(0),\n",
    "    schema=[\n",
    "        Field(name=\"item_category\", dtype=Int32),\n",
    "        Field(name=\"item_shop\", dtype=Int32),\n",
    "        Field(name=\"item_brand\", dtype=Int32),\n",
    "    ],\n",
    "    online=True,\n",
    "    source=item_features,\n",
    "    tags=dict(),\n",
    ")\n",
    "\"\"\".format(\n",
    "            os.path.join(feature_repo_path, \"data/\", \"item_features.parquet\")\n",
    "        )\n",
    "    )\n",
    "file.close()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "660333b2-4f99-49c7-8cd3-f0aad5dbd66f",
   "metadata": {},
   "source": [
    "Let's checkout our Feast feature repository structure."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "id": "57133c1e-18d9-4ccb-9704-cdebd271985e",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Requirement already satisfied: seedir in /usr/local/lib/python3.8/dist-packages (0.4.2)\n",
      "Requirement already satisfied: natsort in /usr/local/lib/python3.8/dist-packages (from seedir) (8.4.0)\n"
     ]
    }
   ],
   "source": [
    "# install seedir\n",
    "!pip install seedir"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "id": "986d53ea-c946-4046-a390-6d3b8801d280",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "feast_repo/\n",
      "├─README.md\n",
      "├─__init__.py\n",
      "└─feature_repo/\n",
      "  ├─__init__.py\n",
      "  ├─__pycache__/\n",
      "  │ ├─__init__.cpython-38.pyc\n",
      "  │ ├─example_repo.cpython-38.pyc\n",
      "  │ └─test_workflow.cpython-38.pyc\n",
      "  ├─data/\n",
      "  │ ├─item_features.parquet\n",
      "  │ └─user_features.parquet\n",
      "  ├─feature_store.yaml\n",
      "  ├─item_features.py\n",
      "  ├─test_workflow.py\n",
      "  └─user_features.py\n"
     ]
    }
   ],
   "source": [
    "import seedir as sd\n",
    "\n",
    "feature_repo_path = os.path.join(BASE_DIR, \"feast_repo\")\n",
    "sd.seedir(\n",
    "    feature_repo_path,\n",
    "    style=\"lines\",\n",
    "    itemlimit=10,\n",
    "    depthlimit=3,\n",
    "    exclude_folders=\".ipynb_checkpoints\",\n",
    "    sort=True,\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "80678ea1-a7fb-4016-9e6f-c905497f4142",
   "metadata": {},
   "source": [
    "### Next Steps\n",
    "We trained and exported our ranking and retrieval models and NVTabular workflows. In the next step, we will learn how to deploy our trained models into [Triton Inference Server (TIS)](https://github.com/triton-inference-server/server) with Merlin Systems library.\n",
    "\n",
    "For the next step, move on to the `02-Deploying-multi-stage-Recsys-with-Merlin-Systems.ipynb` notebook to deploy our saved models as an ensemble to TIS and obtain prediction results for a given request."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3c5bd646-8121-4f32-bff8-137d50e3b8a2",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "interpreter": {
   "hash": "2758ff992bb32b90e83258e2e763c5fcee80c4002721441c6c0d17c649a641dd"
  },
  "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"
  },
  "merlin": {
   "containers": [
    "nvcr.io/nvidia/merlin/merlin-tensorflow-inference:latest"
   ]
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}