![](\"https://developer.download.nvidia.com/notebooks/dlsw%20notebooks/merlin_models-transformers-next-item-prediction-with-pretrained-embeddings/nvidia_logo.png\"){kind=logo}
\n",
+ "\n",
+ "# Transformer-based architecture for next-item prediction task with pretrained embeddings\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.\n",
+ "\n",
+ "## Overview\n",
+ "\n",
+ "In this use case we will train a Transformer-based architecture for next-item prediction task with pretrained embeddings.\n",
+ "\n",
+ "**You can chose to download the full dataset manually or use synthetic data.**\n",
+ "\n",
+ "We will use the [SIGIR eCOM 2021 Data Challenge Dataset](https://github.com/coveooss/SIGIR-ecom-data-challenge) to train a session-based model. The dataset contains 36M events of users browsing an online store.\n",
+ "\n",
+ "We will reshape the data to organize it into 'sessions'. Each session will be a full customer online journey in chronological order. The goal will be to predict the `url` of the next action taken.\n",
+ "\n",
+ "\n",
+ "### Learning objectives\n",
+ "\n",
+ "- Training a Transformer-based architecture for next-item prediction task"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "fd2b847f",
+ "metadata": {},
+ "source": [
+ "## Downloading and preparing the dataset"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": 2,
+ "id": "2dd7827c",
+ "metadata": {},
+ "outputs": [
+ {
+ "name": "stderr",
+ "output_type": "stream",
+ "text": [
+ "2023-06-20 22:58:36.667322: 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",
+ "2023-06-20 22:58:38.026020: I tensorflow/compiler/xla/stream_executor/cuda/cuda_gpu_executor.cc:998] successful NUMA node read from SysFS had negative value (-1), but there must be at least one NUMA node, so returning NUMA node zero\n",
+ "2023-06-20 22:58:38.026445: I tensorflow/compiler/xla/stream_executor/cuda/cuda_gpu_executor.cc:998] successful NUMA node read from SysFS had negative value (-1), but there must be at least one NUMA node, so returning NUMA node zero\n",
+ "2023-06-20 22:58:38.026622: I tensorflow/compiler/xla/stream_executor/cuda/cuda_gpu_executor.cc:998] successful NUMA node read from SysFS had negative value (-1), but there must be at least one NUMA node, so returning NUMA node zero\n"
+ ]
+ }
+ ],
+ "source": [
+ "import os\n",
+ "import cudf\n",
+ "import numpy as np\n",
+ "import pandas as pd\n",
+ "import nvtabular as nvt\n",
+ "from merlin.schema import ColumnSchema, Schema, Tags\n",
+ "\n",
+ "OUTPUT_DATA_DIR = os.environ.get('OUTPUT_DATA_DIR', '/workspace/data')\n",
+ "NUM_EPOCHS = int(os.environ.get('NUM_EPOCHS', 5))\n",
+ "NUM_EXAMPLES = int(os.environ.get('NUM_EXAMPLES', 100_000))\n",
+ "MINIMUM_SESSION_LENGTH = int(os.environ.get('MINIMUM_SESSION_LENGTH', 5))"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "7fcf7c86",
+ "metadata": {},
+ "source": [
+ "You can download the full dataset by registering [here](https://www.coveo.com/en/ailabs/sigir-ecom-data-challenge). If you chose to download the data, please place it alongside this notebook in the `sigir_dataset` directory and extract it.\n",
+ "\n",
+ "By default, in this notebook we will be using synthetically generated data based on the SIGIR dataset, but you can run on the full dataset by changing the value of the boolean flag below."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": 3,
+ "id": "bc3d1882",
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "RUN_ON_SYNTHETIC_DATA = True"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "68bc6d6d",
+ "metadata": {},
+ "source": [
+ "### Clean downloaded data"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "9016a3e2",
+ "metadata": {},
+ "source": [
+ "If you are training on the full SIGIR dataset, the following code will pre-process it.\n",
+ "\n",
+ "Here we deal with `nan` values, drop rows with missing information and parse strings containing lists to lists.\n",
+ "\n",
+ "The synthetically generated data is already clean -- it doesn't require this pre-processing."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": 4,
+ "id": "428ab049",
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "if not RUN_ON_SYNTHETIC_DATA:\n",
+ " train = nvt.Dataset('/workspace/sigir_dataset/train/browsing_train.csv', part_size='500MB')\n",
+ " skus = nvt.Dataset('/workspace/sigir_dataset/train/sku_to_content.csv')\n",
+ "\n",
+ " skus = pd.read_csv('/workspace/sigir_dataset/train/sku_to_content.csv')\n",
+ "\n",
+ " skus['description_vector'] = skus['description_vector'].replace(np.nan, '')\n",
+ " skus['image_vector'] = skus['image_vector'].replace(np.nan, '')\n",
+ "\n",
+ " skus['description_vector'] = skus['description_vector'].apply(lambda x: [] if len(x) == 0 else eval(x))\n",
+ " skus['image_vector'] = skus['image_vector'].apply(lambda x: [] if len(x) == 0 else eval(x))\n",
+ " skus = skus[skus.description_vector.apply(len) > 0]\n",
+ " skus = nvt.Dataset(skus)"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "9b33fa32",
+ "metadata": {},
+ "source": [
+ "### Generate synthetic data"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "4c4ba9b9",
+ "metadata": {},
+ "source": [
+ "If you are not running on the full dataset, the following lines of code will generate its synthetic counterpart."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": 5,
+ "id": "84789211",
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "if RUN_ON_SYNTHETIC_DATA:\n",
+ " from merlin.datasets.synthetic import generate_data\n",
+ "\n",
+ " train = generate_data('sigir-browsing', NUM_EXAMPLES)\n",
+ " skus = generate_data('sigir-sku', NUM_EXAMPLES)"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "5533f446",
+ "metadata": {},
+ "source": [
+ "## Constructing a workflow"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "ac47bc4e",
+ "metadata": {},
+ "source": [
+ "We need to process our data further before we can use it to train our model.\n",
+ "\n",
+ "In particular, the `skus` dataset contains the mapping between the `product_sku_hash` (essentially an item id) to the `description_vector` -- an embedding obtained from the description.\n",
+ "\n",
+ "We would like to enable our model to make use of this piece of information. In order to feed this data to our model, we need to map the `product_sku_hash` to an id.\n",
+ "\n",
+ "But we need to make sure that the way we process `skus` and the `train` dataset (event information) is consistent, that the same `product_sku_hash` is mapped to the same id both when processing `skus` and `train`.\n",
+ "\n",
+ "We do so by defining and fitting a `Categorify` op once and using it to process both the `skus` and the `train` datasets.\n",
+ "\n",
+ "Additionally, we apply some further processing to the `train` dataset. We group rows by `session_id_hash` so that each training example will contain events from a single customer visit to the online store arranged in chronological order.\n",
+ "\n",
+ "If you would like to learn more about leveraging `NVTabular` to process tabular data on the GPU using a set of industry standard operators, please consult the examples available [here](https://github.com/NVIDIA-Merlin/NVTabular/tree/main/examples).\n",
+ "\n",
+ "Let's first process the `train` dataset and retain the `Categorify` operator (`cat_op`) for processing of `skus`."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": 6,
+ "id": "3b5feee3",
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "cat_op = nvt.ops.Categorify()\n",
+ "out = ['product_sku_hash'] >> cat_op >> nvt.ops.TagAsItemID()\n",
+ "out += ['event_type', 'product_action', 'session_id_hash', 'hashed_url'] >> nvt.ops.Categorify()\n",
+ "out += ['server_timestamp_epoch_ms'] >> nvt.ops.NormalizeMinMax()\n",
+ "\n",
+ "groupby_features = out >> nvt.ops.Groupby(\n",
+ " groupby_cols=['session_id_hash'],\n",
+ " aggs={\n",
+ " 'product_sku_hash': ['list'],\n",
+ " 'event_type': ['list'],\n",
+ " 'product_action': ['list'],\n",
+ " 'hashed_url': ['list', 'count'],\n",
+ " 'server_timestamp_epoch_ms': ['list']\n",
+ " },\n",
+ " sort_cols=\"server_timestamp_epoch_ms\"\n",
+ ")\n",
+ "\n",
+ "filtered_sessions = groupby_features >> nvt.ops.Filter(f=lambda df: df[\"hashed_url_count\"] >= MINIMUM_SESSION_LENGTH)\n",
+ "\n",
+ "# We won't be needing the `session_id_hash` nor the `hashed_url_count` any longer\n",
+ "wf = nvt.Workflow(\n",
+ " filtered_sessions[\n",
+ " 'product_sku_hash_list',\n",
+ " 'event_type_list',\n",
+ " 'product_action_list',\n",
+ " 'hashed_url_list',\n",
+ " ]\n",
+ ")\n",
+ "\n",
+ "# Let's save the output of our workflow -- transformed `train` for later use (training of our model).\n",
+ "wf.fit_transform(train).to_parquet('train_transformed')"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "45a4828e",
+ "metadata": {},
+ "source": [
+ "Here are a couple of example rows from `train_transformed`."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": 7,
+ "id": "650fb0d0",
+ "metadata": {},
+ "outputs": [
+ {
+ "data": {
+ "text/html": [
+ "\n", + "I0414 18:29:50.742197 4067 http_server.cc:3113] Started HTTPService at 0.0.0.0:8000
\n", + "I0414 18:29:50.783470 4067 http_server.cc:178] Started Metrics Service at 0.0.0.0:8002\n", + "\n", + "Let us now package our data for inference. We will send 5 rows of data, which corresponds to a single customer journey (session) through the online store. The data will be first processed by the `NVTabular` workflow and subsequentally passed to our transformer model for predicting. " + ] + }, + { + "cell_type": "code", + "execution_count": 24, + "id": "90483210", + "metadata": {}, + "outputs": [], + "source": [ + "# obtaining five rows of data\n", + "df = train.head(5)\n", + "# making sure all the rows correspond to the same online session (have the same `session_id_hash`)\n", + "df['session_id_hash'] = df['session_id_hash'].iloc[0]" + ] + }, + { + "cell_type": "markdown", + "id": "efdf671e", + "metadata": {}, + "source": [ + "Let us now send the data to the Triton Inference Server for inference." + ] + }, + { + "cell_type": "code", + "execution_count": 25, + "id": "d8453048", + "metadata": {}, + "outputs": [], + "source": [ + "from merlin.systems.triton import convert_df_to_triton_input\n", + "import tritonclient.grpc as grpcclient\n", + "\n", + "inputs = convert_df_to_triton_input(wf.input_schema, df)\n", + "\n", + "with grpcclient.InferenceServerClient(\"localhost:8001\") as client:\n", + " response = client.infer('executor_model', inputs)" + ] + }, + { + "cell_type": "markdown", + "id": "913b80e8", + "metadata": {}, + "source": [ + "Let's parse the response." + ] + }, + { + "cell_type": "code", + "execution_count": 26, + "id": "4cc4b046", + "metadata": {}, + "outputs": [ + { + "data": { + "text/plain": [ + "array([[-2.2332087 , -2.1218574 , -2.390479 , ..., -0.7735352 ,\n", + " 0.1954267 , -0.34523243]], dtype=float32)" + ] + }, + "execution_count": 26, + "metadata": {}, + "output_type": "execute_result" + } + ], + "source": [ + "predictions = response.as_numpy(\"hashed_url_list/categorical_output\")\n", + "predictions" + ] + }, + { + "cell_type": "markdown", + "id": "e49c2ed9", + "metadata": {}, + "source": [ + "The response contains logits predicting the id of the url the customer is most likely to arrive at as next step of their journey through the online store.\n", + "\n", + "Here is the predicted hashed url id:" + ] + }, + { + "cell_type": "code", + "execution_count": 27, + "id": "0b9af2ae", + "metadata": {}, + "outputs": [ + { + "data": { + "text/plain": [ + "34" + ] + }, + "execution_count": 27, + "metadata": {}, + "output_type": "execute_result" + } + ], + "source": [ + "predicted_hashed_url_id = predictions.argmax()\n", + "predicted_hashed_url_id" + ] + }, + { + "cell_type": "markdown", + "id": "8ef47efd", + "metadata": {}, + "source": [ + "## Summary\n", + "\n", + "We have trained a transformer model for the next item prediction task using language model masking.\n", + "\n", + "For another session-based example that goes deeper into data preprocessing and that covers several advanced techniques (Weight Tying, Temperature Scaling) please see [Session-Based Next Item Prediction for Fashion E-Commerce](https://github.com/NVIDIA-Merlin/models/blob/t4rec_use_case/examples/usecases/ecommerce-session-based-next-item-prediction-for-fashion.ipynb). " + ] + } + ], + "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" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +} diff --git a/examples/Next-Item-Prediction-with-Transformers/tf/transformers-next-item-prediction.ipynb b/examples/Next-Item-Prediction-with-Transformers/tf/transformers-next-item-prediction.ipynb new file mode 100644 index 000000000..b409170f0 --- /dev/null +++ b/examples/Next-Item-Prediction-with-Transformers/tf/transformers-next-item-prediction.ipynb @@ -0,0 +1,1516 @@ +{ + "cells": [ + { + "cell_type": "code", + "execution_count": 1, + "id": "a556f660", + "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 anda\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": "697d1452", + "metadata": {}, + "source": [ + "
![](\"https://developer.download.nvidia.com/notebooks/dlsw-notebooks/merlin_models-transformers-net-item-prediction/nvidia_logo.png\"){kind=logo}
\n",
+ "\n",
+ "# Transformer-based architecture for next-item prediction task\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.\n",
+ "\n",
+ "## Overview\n",
+ "\n",
+ "In this use case we will train a Transformer-based architecture for next-item prediction task.\n",
+ "\n",
+ "**Note, the data for this notebook will be automatically downloaded to the folder specified in the cells below.**\n",
+ "\n",
+ "We will use the [booking.com dataset](https://github.com/bookingcom/ml-dataset-mdt) to train a session-based model. The dataset contains 1,166,835 of anonymized hotel reservations in the train set and 378,667 in the test set. Each reservation is a part of a customer's trip (identified by `utrip_id`) which includes consecutive reservations.\n",
+ "\n",
+ "We will reshape the data to organize it into 'sessions'. Each session will be a full customer itinerary in chronological order. The goal will be to predict the city_id of the final reservation of each trip.\n",
+ "\n",
+ "\n",
+ "### Learning objectives\n",
+ "\n",
+ "- Training a Transformer-based architecture for next-item prediction task"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "1cccd005",
+ "metadata": {},
+ "source": [
+ "## Downloading and preparing the dataset"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "1d0b619b",
+ "metadata": {},
+ "source": [
+ "We will download the dataset using a functionality provided by merlin models. The dataset can be found on GitHub [here](https://github.com/bookingcom/ml-dataset-mdt).\n",
+ "\n",
+ "**Read more about libraries used in the import statements below**\n",
+ "\n",
+ "- [get_lib](https://github.com/NVIDIA-Merlin/core/blob/stable/merlin/core/dispatch.py)\n",
+ "- [get_booking](https://github.com/NVIDIA-Merlin/models/tree/stable/merlin/datasets/ecommerce)\n",
+ "- [nvtabular](https://github.com/NVIDIA-Merlin/NVTabular/tree/stable/nvtabular)\n",
+ "- [nvtabular ops](https://github.com/NVIDIA-Merlin/NVTabular/tree/stable/nvtabular/ops)\n",
+ "- [schema tags](https://github.com/NVIDIA-Merlin/core/blob/stable/merlin/schema/tags.py)\n",
+ "- [merlin models tensorflow](https://github.com/NVIDIA-Merlin/models/tree/stable/merlin/models/tf)\n",
+ "- [get_booking](https://github.com/NVIDIA-Merlin/models/blob/stable/merlin/datasets/ecommerce/booking/dataset.py)"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": 2,
+ "id": "40e9ef05",
+ "metadata": {},
+ "outputs": [
+ {
+ "name": "stderr",
+ "output_type": "stream",
+ "text": [
+ "2023-05-31 06:06:25.697025: 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"
+ ]
+ },
+ {
+ "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"
+ ]
+ },
+ {
+ "name": "stderr",
+ "output_type": "stream",
+ "text": [
+ "/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",
+ "2023-05-31 06:06:26.988036: I tensorflow/compiler/xla/stream_executor/cuda/cuda_gpu_executor.cc:998] successful NUMA node read from SysFS had negative value (-1), but there must be at least one NUMA node, so returning NUMA node zero\n",
+ "2023-05-31 06:06:26.988386: I tensorflow/compiler/xla/stream_executor/cuda/cuda_gpu_executor.cc:998] successful NUMA node read from SysFS had negative value (-1), but there must be at least one NUMA node, so returning NUMA node zero\n",
+ "2023-05-31 06:06:26.988518: I tensorflow/compiler/xla/stream_executor/cuda/cuda_gpu_executor.cc:998] successful NUMA node read from SysFS had negative value (-1), but there must be at least one NUMA node, so returning NUMA node zero\n"
+ ]
+ },
+ {
+ "name": "stdout",
+ "output_type": "stream",
+ "text": [
+ "[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",
+ "[SOK INFO] Initialize finished, communication tool: horovod\n"
+ ]
+ },
+ {
+ "name": "stderr",
+ "output_type": "stream",
+ "text": [
+ "2023-05-31 06:06:28.519868: 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-05-31 06:06:28.520815: I tensorflow/compiler/xla/stream_executor/cuda/cuda_gpu_executor.cc:998] successful NUMA node read from SysFS had negative value (-1), but there must be at least one NUMA node, so returning NUMA node zero\n",
+ "2023-05-31 06:06:28.520999: I tensorflow/compiler/xla/stream_executor/cuda/cuda_gpu_executor.cc:998] successful NUMA node read from SysFS had negative value (-1), but there must be at least one NUMA node, so returning NUMA node zero\n",
+ "2023-05-31 06:06:28.521129: I tensorflow/compiler/xla/stream_executor/cuda/cuda_gpu_executor.cc:998] successful NUMA node read from SysFS had negative value (-1), but there must be at least one NUMA node, so returning NUMA node zero\n",
+ "2023-05-31 06:06:28.591345: I tensorflow/compiler/xla/stream_executor/cuda/cuda_gpu_executor.cc:998] successful NUMA node read from SysFS had negative value (-1), but there must be at least one NUMA node, so returning NUMA node zero\n",
+ "2023-05-31 06:06:28.591534: I tensorflow/compiler/xla/stream_executor/cuda/cuda_gpu_executor.cc:998] successful NUMA node read from SysFS had negative value (-1), but there must be at least one NUMA node, so returning NUMA node zero\n",
+ "2023-05-31 06:06:28.591665: I tensorflow/compiler/xla/stream_executor/cuda/cuda_gpu_executor.cc:998] successful NUMA node read from SysFS had negative value (-1), but there must be at least one NUMA node, so returning NUMA node zero\n",
+ "2023-05-31 06:06:28.591770: 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-05-31 06:06:28.591778: I tensorflow/core/common_runtime/gpu/gpu_process_state.cc:222] Using CUDA malloc Async allocator for GPU: 0\n",
+ "2023-05-31 06:06:28.591860: 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:08: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"
+ ]
+ }
+ ],
+ "source": [
+ "# Resetting the TF memory allocation to not be 50% by default. \n",
+ "import os\n",
+ "os.environ[\"TF_GPU_ALLOCATOR\"]=\"cuda_malloc_async\"\n",
+ "\n",
+ "from merlin.core.dispatch import get_lib\n",
+ "from merlin.datasets.ecommerce import get_booking\n",
+ "\n",
+ "import numpy as np\n",
+ "import timeit\n",
+ "\n",
+ "from nvtabular import *\n",
+ "from nvtabular import ops\n",
+ "\n",
+ "from merlin.schema.tags import Tags\n",
+ "import merlin.models.tf as mm\n",
+ "\n",
+ "INPUT_DATA_DIR = os.environ.get('INPUT_DATA_DIR', '/workspace/data')\n",
+ "OUTPUT_DATA_DIR = os.environ.get('OUTPUT_DATA_DIR', '/workspace/data')\n",
+ "NUM_EPOCHS = int(os.environ.get('NUM_EPOCHS', '5'))"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "c1b42076",
+ "metadata": {},
+ "source": [
+ "Let's download the data."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": 3,
+ "id": "d0a33352",
+ "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.USER_ID have been deprecated and will be removed in a future version. Please use the atomic versions of these tags, like [\n",
+ "\n",
+ "\n",
+ " \n",
+ "
\n",
+ "
"
+ ],
+ "text/plain": [
+ " city_id_list booker_country_list \\\n",
+ "0 [8238, 156, 2278, 2097] [3, 3, 3, 3] \n",
+ "1 [63, 1160, 87, 618, 63] [1, 1, 1, 1, 1] \n",
+ "2 [7, 6, 24, 1050, 65, 52, 3] [2, 2, 2, 2, 2, 2, 2] \n",
+ "3 [1032, 757, 140, 3] [2, 2, 2, 2] \n",
+ "4 [3603, 262, 662, 250, 359] [1, 1, 1, 1, 1] \n",
+ "\n",
+ " hotel_country_list weekday_checkin_list weekday_checkout_list \\\n",
+ "0 [3, 3, 3, 3] [5, 7, 4, 3] [7, 4, 2, 7] \n",
+ "1 [1, 1, 1, 1, 1] [5, 1, 4, 3, 5] [6, 4, 2, 5, 4] \n",
+ "2 [2, 2, 2, 16, 16, 3, 3] [5, 1, 2, 6, 5, 7, 4] [6, 3, 1, 5, 7, 4, 3] \n",
+ "3 [19, 19, 19, 3] [1, 4, 2, 3] [4, 3, 2, 5] \n",
+ "4 [30, 30, 30, 30, 30] [1, 3, 6, 5, 1] [2, 1, 5, 6, 3] \n",
+ "\n",
+ " city_id_count \n",
+ "0 4 \n",
+ "1 5 \n",
+ "2 7 \n",
+ "3 4 \n",
+ "4 5 "
+ ]
+ },
+ "execution_count": 10,
+ "metadata": {},
+ "output_type": "execute_result"
+ }
+ ],
+ "source": [
+ "Dataset(os.path.join(OUTPUT_DATA_DIR, 'train_processed.parquet')).head()"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "e89cc3a0",
+ "metadata": {},
+ "source": [
+ "We are now ready to train our model."
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "ce95c794",
+ "metadata": {},
+ "source": [
+ "Here is the schema of the data that our model will use."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": 11,
+ "id": "c4813456",
+ "metadata": {},
+ "outputs": [
+ {
+ "data": {
+ "text/html": [
+ "| \n", + " | city_id_list | \n", + "booker_country_list | \n", + "hotel_country_list | \n", + "weekday_checkin_list | \n", + "weekday_checkout_list | \n", + "city_id_count | \n", + "
|---|---|---|---|---|---|---|
| 0 | \n", + "[8238, 156, 2278, 2097] | \n", + "[3, 3, 3, 3] | \n", + "[3, 3, 3, 3] | \n", + "[5, 7, 4, 3] | \n", + "[7, 4, 2, 7] | \n", + "4 | \n", + "
| 1 | \n", + "[63, 1160, 87, 618, 63] | \n", + "[1, 1, 1, 1, 1] | \n", + "[1, 1, 1, 1, 1] | \n", + "[5, 1, 4, 3, 5] | \n", + "[6, 4, 2, 5, 4] | \n", + "5 | \n", + "
| 2 | \n", + "[7, 6, 24, 1050, 65, 52, 3] | \n", + "[2, 2, 2, 2, 2, 2, 2] | \n", + "[2, 2, 2, 16, 16, 3, 3] | \n", + "[5, 1, 2, 6, 5, 7, 4] | \n", + "[6, 3, 1, 5, 7, 4, 3] | \n", + "7 | \n", + "
| 3 | \n", + "[1032, 757, 140, 3] | \n", + "[2, 2, 2, 2] | \n", + "[19, 19, 19, 3] | \n", + "[1, 4, 2, 3] | \n", + "[4, 3, 2, 5] | \n", + "4 | \n", + "
| 4 | \n", + "[3603, 262, 662, 250, 359] | \n", + "[1, 1, 1, 1, 1] | \n", + "[30, 30, 30, 30, 30] | \n", + "[1, 3, 6, 5, 1] | \n", + "[2, 1, 5, 6, 3] | \n", + "5 | \n", + "
\n",
+ "\n",
+ "\n",
+ " \n",
+ "
\n",
+ "
"
+ ],
+ "text/plain": [
+ "[{'name': 'city_id_list', 'tags': {| \n", + " | name | \n", + "tags | \n", + "dtype | \n", + "is_list | \n", + "is_ragged | \n", + "properties.num_buckets | \n", + "properties.freq_threshold | \n", + "properties.max_size | \n", + "properties.start_index | \n", + "properties.cat_path | \n", + "properties.domain.min | \n", + "properties.domain.max | \n", + "properties.domain.name | \n", + "properties.embedding_sizes.cardinality | \n", + "properties.embedding_sizes.dimension | \n", + "properties.value_count.min | \n", + "properties.value_count.max | \n", + "
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| 0 | \n", + "city_id_list | \n", + "(Tags.SEQUENCE, Tags.CATEGORICAL) | \n", + "DType(name='int64', element_type=<ElementType.... | \n", + "True | \n", + "True | \n", + "None | \n", + "0 | \n", + "0 | \n", + "0 | \n", + ".//categories/unique.city_id.parquet | \n", + "0 | \n", + "37202 | \n", + "city_id | \n", + "37203 | \n", + "512 | \n", + "0 | \n", + "None | \n", + "
| 1 | \n", + "booker_country_list | \n", + "(Tags.SEQUENCE, Tags.CATEGORICAL) | \n", + "DType(name='int64', element_type=<ElementType.... | \n", + "True | \n", + "True | \n", + "None | \n", + "0 | \n", + "0 | \n", + "0 | \n", + ".//categories/unique.booker_country.parquet | \n", + "0 | \n", + "5 | \n", + "booker_country | \n", + "6 | \n", + "16 | \n", + "0 | \n", + "None | \n", + "
| 2 | \n", + "hotel_country_list | \n", + "(Tags.SEQUENCE, Tags.CATEGORICAL) | \n", + "DType(name='int64', element_type=<ElementType.... | \n", + "True | \n", + "True | \n", + "None | \n", + "0 | \n", + "0 | \n", + "0 | \n", + ".//categories/unique.hotel_country.parquet | \n", + "0 | \n", + "194 | \n", + "hotel_country | \n", + "195 | \n", + "31 | \n", + "0 | \n", + "None | \n", + "
| 3 | \n", + "weekday_checkin_list | \n", + "(Tags.SEQUENCE, Tags.CATEGORICAL) | \n", + "DType(name='int64', element_type=<ElementType.... | \n", + "True | \n", + "True | \n", + "None | \n", + "0 | \n", + "0 | \n", + "0 | \n", + ".//categories/unique.weekday_checkin.parquet | \n", + "0 | \n", + "7 | \n", + "weekday_checkin | \n", + "8 | \n", + "16 | \n", + "0 | \n", + "None | \n", + "
| 4 | \n", + "weekday_checkout_list | \n", + "(Tags.SEQUENCE, Tags.CATEGORICAL) | \n", + "DType(name='int64', element_type=<ElementType.... | \n", + "True | \n", + "True | \n", + "None | \n", + "0 | \n", + "0 | \n", + "0 | \n", + ".//categories/unique.weekday_checkout.parquet | \n", + "0 | \n", + "7 | \n", + "weekday_checkout | \n", + "8 | \n", + "16 | \n", + "0 | \n", + "None | \n", + "
\n", + "I0414 18:29:50.742197 4067 http_server.cc:3113] Started HTTPService at 0.0.0.0:8000
\n", + "I0414 18:29:50.783470 4067 http_server.cc:178] Started Metrics Service at 0.0.0.0:8002\n", + "\n", + "Let us now package our data for inference. We will send the first 4 rows of our validation data, which corresponds to a single trip. The data will be first processed by the `NVTabular` workflow and subsequentally passed to our transformer model for predicting. " + ] + }, + { + "cell_type": "markdown", + "id": "d83a304d", + "metadata": {}, + "source": [ + "Let us send the first 4 rows of our validation data to Triton. This will correspond to a single trip (all rows have the same `utrip_id`) with four stops." + ] + }, + { + "cell_type": "code", + "execution_count": 18, + "id": "3cad9026", + "metadata": {}, + "outputs": [], + "source": [ + "from merlin.systems.triton import convert_df_to_triton_input\n", + "\n", + "validation_data = validation_set_dataset.compute()\n", + "inputs = convert_df_to_triton_input(wf.input_schema, validation_data.iloc[:4])" + ] + }, + { + "cell_type": "code", + "execution_count": 19, + "id": "c508adce", + "metadata": {}, + "outputs": [], + "source": [ + "import tritonclient.grpc as grpcclient\n", + "\n", + "with grpcclient.InferenceServerClient(\"localhost:8001\") as client:\n", + " response = client.infer('executor_model', inputs)" + ] + }, + { + "cell_type": "markdown", + "id": "6d34eecf", + "metadata": {}, + "source": [ + "The response consists of logits coming from our model." + ] + }, + { + "cell_type": "code", + "execution_count": 20, + "id": "b3284691", + "metadata": {}, + "outputs": [ + { + "data": { + "text/plain": [ + "array([[-2.8206294 , -1.3849059 , 1.9042726 , ..., 0.851537 ,\n", + " -2.4237087 , -0.73849726]], dtype=float32)" + ] + }, + "execution_count": 20, + "metadata": {}, + "output_type": "execute_result" + } + ], + "source": [ + "response.as_numpy('city_id_list/categorical_output')" + ] + }, + { + "cell_type": "code", + "execution_count": 21, + "id": "824d2b4f", + "metadata": {}, + "outputs": [ + { + "data": { + "text/plain": [ + "(1, 37203)" + ] + }, + "execution_count": 21, + "metadata": {}, + "output_type": "execute_result" + } + ], + "source": [ + "predictions = response.as_numpy('city_id_list/categorical_output')\n", + "predictions.shape" + ] + }, + { + "cell_type": "markdown", + "id": "fc5d415b", + "metadata": {}, + "source": [ + "The above values are logits output from the last layer of our model. They correspond in size to the cardinality of `city_id`, our target variable:" + ] + }, + { + "cell_type": "code", + "execution_count": 22, + "id": "29a8c0bd", + "metadata": {}, + "outputs": [ + { + "data": { + "text/plain": [ + "37203" + ] + }, + "execution_count": 22, + "metadata": {}, + "output_type": "execute_result" + } + ], + "source": [ + "cardinality = wf.output_schema['city_id_list'].properties['embedding_sizes']['cardinality']\n", + "cardinality" + ] + }, + { + "cell_type": "markdown", + "id": "3c54c30f", + "metadata": {}, + "source": [ + "## Summary" + ] + }, + { + "cell_type": "markdown", + "id": "709c07fb", + "metadata": {}, + "source": [ + "We have trained a transformer model for the next item prediction task using language model masking.\n", + "\n", + "For another session-based example that goes deeper into data preprocessing and that covers several advanced techniques (Weight Tying, Temperature Scaling) please see [Session-Based Next Item Prediction for Fashion E-Commerce](https://github.com/NVIDIA-Merlin/models/blob/t4rec_use_case/examples/usecases/ecommerce-session-based-next-item-prediction-for-fashion.ipynb). " + ] + } + ], + "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" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +} diff --git a/pytest.ini b/pytest.ini index 3b9940dfb..09508941f 100644 --- a/pytest.ini +++ b/pytest.ini @@ -2,3 +2,4 @@ markers = multigpu: Tests only run in multiple-GPU environments singlegpu: Optional marker to run tests in single-GPU environments. Usually used when running in both single- and multi-GPU. + notebook: mark as testing notebooks diff --git a/tests/benchmark/test_asvdb_transformers_next_item_prediction.py b/tests/benchmark/test_asvdb_transformers_next_item_prediction.py new file mode 100644 index 000000000..ccd21015a --- /dev/null +++ b/tests/benchmark/test_asvdb_transformers_next_item_prediction.py @@ -0,0 +1,95 @@ +from asvdb import ASVDb, BenchmarkResult, utils +from testbook import testbook + +from tests.conftest import REPO_ROOT, get_benchmark_info + + +@testbook( + REPO_ROOT / "examples/Next-Item-Prediction-with-Transformers/tf/transformers-next-item-prediction.ipynb", + timeout=720, + execute=False, +) +def test_func(tb, tmpdir): + tb.inject( + f""" + import os + os.environ["INPUT_DATA_DIR"] = "/raid/data/booking" + os.environ["OUTPUT_DATA_DIR"] = "{tmpdir}" + os.environ["NUM_EPOCHS"] = '1' + """ + ) + tb.cells.pop(6) + tb.cells[ + 15 + ].source = """ + def process_data(): + wf = Workflow(filtered_sessions) + + wf.fit_transform(train_set_dataset).to_parquet( + os.path.join(OUTPUT_DATA_DIR, 'train_processed.parquet') + ) + wf.transform(validation_set_dataset).to_parquet( + os.path.join(OUTPUT_DATA_DIR, 'validation_processed.parquet') + ) + + wf.save(os.path.join(OUTPUT_DATA_DIR, 'workflow')) + + data_processing_runtime = timeit.timeit(process_data, number=1) + """ + tb.cells[ + 29 + ].source = """ + model.compile(run_eagerly=False, optimizer='adam', loss="categorical_crossentropy") + + def train_model(): + model.fit( + Dataset(os.path.join(OUTPUT_DATA_DIR, 'train_processed.parquet')), + batch_size=64, + epochs=NUM_EPOCHS, + pre=mm.SequenceMaskRandom( + schema=seq_schema, + target=target, + masking_prob=0.3, + transformer=transformer_block + ) + ) + + training_runtime = timeit.timeit(train_model, number=1) + """ + tb.execute_cell(list(range(0, 35))) + data_processing_runtime = tb.ref("data_processing_runtime") + training_runtime = tb.ref("training_runtime") + ndcg_at_10 = tb.ref("metrics")["ndcg_at_10"] + + bResult1 = BenchmarkResult( + funcName="", + argNameValuePairs=[ + ("notebook_name", "usecases/transformers-next-item-prediction"), + ("measurement", "data_processing_runtime"), + ], + result=data_processing_runtime, + ) + bResult2 = BenchmarkResult( + funcName="", + argNameValuePairs=[ + ("notebook_name", "usecases/transformers-next-item-prediction"), + ("measurement", "training_runtime"), + ], + result=training_runtime, + ) + bResult3 = BenchmarkResult( + funcName="", + argNameValuePairs=[ + ("notebook_name", "usecases/transformers-next-item-prediction"), + ("measurement", "ndcg_at_10"), + ], + result=ndcg_at_10, + ) + + bInfo = get_benchmark_info() + (repo, branch) = utils.getRepoInfo() + + db = ASVDb(dbDir="s3://nvtab-bench-asvdb/models_metric_tracking", repo=repo, branches=[branch]) + db.addResult(bInfo, bResult1) + db.addResult(bInfo, bResult2) + db.addResult(bInfo, bResult3) diff --git a/tests/conftest.py b/tests/conftest.py index 470c56c40..f05d89f92 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -22,6 +22,16 @@ from pathlib import Path import pytest +import platform +import warnings +from pathlib import Path + +import distributed +import psutil +import pytest +from asvdb import BenchmarkInfo, utils + +from merlin.models.utils import ci_utils REPO_ROOT = Path(__file__).parent.parent @@ -62,3 +72,20 @@ def get_cuda_cluster(): cluster = LocalCUDACluster(n_workers=n_workers) yield cluster cluster.close() + +def get_benchmark_info(): + uname = platform.uname() + (commitHash, commitTime) = utils.getCommitInfo() + + return BenchmarkInfo( + machineName=uname.machine, + cudaVer="na", + osType="%s %s" % (uname.system, uname.release), + pythonVer=platform.python_version(), + commitHash=commitHash, + commitTime=commitTime, + gpuType="na", + cpuType=uname.processor, + arch=uname.machine, + ram="%d" % psutil.virtual_memory().total, + ) diff --git a/tests/unit/test_transformers_next_item_prediction.py b/tests/unit/test_transformers_next_item_prediction.py new file mode 100644 index 000000000..5fe7f31f8 --- /dev/null +++ b/tests/unit/test_transformers_next_item_prediction.py @@ -0,0 +1,57 @@ +import shutil + +import pytest +from testbook import testbook + +from tests.conftest import REPO_ROOT + +pytest.importorskip("transformers") +utils = pytest.importorskip("merlin.systems.triton.utils") + +TRITON_SERVER_PATH = shutil.which("tritonserver") + + +@pytest.mark.skipif(not TRITON_SERVER_PATH, reason="triton server not found") +@testbook( + REPO_ROOT / "examples/Next-Item-Prediction-with-Transformers/tf/transformers-next-item-prediction.ipynb", + timeout=720, + execute=False, +) +@pytest.mark.notebook +def test_next_item_prediction(tb, tmpdir): + tb.inject( + f""" + import os, random + os.environ["INPUT_DATA_DIR"] = "{tmpdir}" + os.environ["OUTPUT_DATA_DIR"] = "{tmpdir}" + from datetime import datetime, timedelta + from merlin.datasets.synthetic import generate_data + ds = generate_data('booking.com-raw', 10000) + df = ds.compute() + def generate_date(): + date = datetime.today() + if random.randint(0, 1): + date -= timedelta(days=7) + return date + df['checkin'] = [generate_date() for _ in range(df.shape[0])] + df['checkout'] = [generate_date() for _ in range(df.shape[0])] + df.to_csv('{tmpdir}/train_set.csv') + """ + ) + tb.cells.pop(6) + tb.cells[29].source = tb.cells[29].source.replace("epochs=5", "epochs=1") + tb.execute_cell(list(range(0, 38))) + + with utils.run_triton_server(f"{tmpdir}/ensemble", grpc_port=8001): + tb.execute_cell(list(range(38, len(tb.cells)))) + + tb.inject( + """ + logits_count = predictions.shape[1] + """ + ) + tb.execute_cell(len(tb.cells) - 1) + + cardinality = tb.ref("cardinality") + logits_count = tb.ref("logits_count") + assert logits_count == cardinality diff --git a/tests/unit/test_transformers_next_item_prediction_with_pretrained_embeddings.py b/tests/unit/test_transformers_next_item_prediction_with_pretrained_embeddings.py new file mode 100644 index 000000000..f3157d1e1 --- /dev/null +++ b/tests/unit/test_transformers_next_item_prediction_with_pretrained_embeddings.py @@ -0,0 +1,38 @@ +import shutil + +import pytest +from testbook import testbook + +from tests.conftest import REPO_ROOT + +pytest.importorskip("transformers") +utils = pytest.importorskip("merlin.systems.triton.utils") + +TRITON_SERVER_PATH = shutil.which("tritonserver") + + +@pytest.mark.skipif(not TRITON_SERVER_PATH, reason="triton server not found") +@testbook( + REPO_ROOT + / "examples/Next-Item-Prediction-with-Transformers/tf/transformers-next-item-prediction-with-pretrained-embeddings.ipynb", + timeout=720, + execute=False, +) +@pytest.mark.notebook +def test_next_item_prediction(tb, tmpdir): + tb.inject( + f""" + import os, random + os.environ["OUTPUT_DATA_DIR"] = "{tmpdir}" + os.environ["NUM_EPOCHS"] = "1" + os.environ["NUM_EXAMPLES"] = "1_500" + os.environ["MINIMUM_SESSION_LENGTH"] = "2" + """ + ) + tb.execute_cell(list(range(0, 48))) + + with utils.run_triton_server(f"{tmpdir}/ensemble", grpc_port=8001): + tb.execute_cell(list(range(48, len(tb.cells)))) + + predicted_hashed_url_id = tb.ref("predicted_hashed_url_id").item() + assert predicted_hashed_url_id >= 0 and predicted_hashed_url_id <= 1002