book-recommender-kubeflow.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "kernelspec": {
      "display_name": "Python 3",
      "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.6.9"
    },
    "colab": {
      "name": "2021-07-01-book-recommender-kubeflow.ipynb",
      "provenance": [],
      "toc_visible": true
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "sfz7gsZ-gWWw"
      },
      "source": [
        "# Books recommendations with Kubeflow Pipelines on Scaleway Kapsule\n",
        "> A simple book recommender system using Kubeflow pipeline\n",
        "\n",
        "- toc: true\n",
        "- badges: true\n",
        "- comments: true\n",
        "- categories: [Book, Kubeflow]\n",
        "- image:"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "iAA9KHclgWW0"
      },
      "source": [
        "**Code examples based on \"[CPU or GPU for your recommendation engine ?](https://blog.scaleway.com/2020/cpu-or-gpu-for-your-recommendation-engine/)\" blogpost by Olga Petrova**\n",
        "\n",
        "Kubeflow Pipelines notebook, by Fabien Da Silva"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "TD4Z-jocgWW1"
      },
      "source": [
        "**Goal**: In this example we are going to learn how to:\n",
        "\n",
        "\n",
        "- Create a Pipeline with 6 tasks \n",
        "  - Download the Dataset\n",
        "  - Prepare the Dataset\n",
        "  - Train a model using Sci-kit Learn (NearestNeighbors)\n",
        "  - Train a model using Pytorch (GPU)\n",
        "  \n",
        "- Use a Persistent Block Storage Volume with Kubeflow Pipelines via a NFS server (store/share datasets, models, ..)\n",
        "\n",
        "- Use GPU efficiently thanks to the Kubeflow engine and Kapsule Auto Scaling"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Kolahy17gWW2"
      },
      "source": [
        "**Prerequisites to run this notebook**:\n",
        " - Have a Kapsule cluster (At least, with one or more GP1-M instance. See Installation guide.\n",
        " - Kubeflow is deployed on tke Kapsule cluster\n",
        " - A GPU node pool is available (you can use Auto-Scaling from 0 to n nodes)\n",
        " - A NFS server is deployed in the Kubeflow Namespace. This will provide a persistent storage on Block Storage, and that can be accessed simultaneously by several Kubernetes Pods (several pipelines)\n",
        " \n",
        " A Scaleway tutorial will be available shortly to describe the installation process (By the time, a preliminary document is provided with this code)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "BdoqI-krgWW3"
      },
      "source": [
        "**Usefull documentation**:\n",
        "\n",
        "- [Kubeflow home page](https://kubeflow.org)\n",
        "- [Kubeflow examples](https://github.com/kubeflow/examples)\n",
        "- [Scaleway Kapsule Product page](https://www.scaleway.com/en/kubernetes-kapsule)\n",
        "- [Kapsule documentation](https://www.scaleway.com/en/docs/get-started-with-scaleway-kubernetes-kapsule/)\n",
        "- [Kubernetes home page](https://kubernetes.io/)\n",
        "- [Scaleway GPU Instances Product page](https://www.scaleway.com/en/gpu-instances/)\n",
        "- [Scaleway Container Registry Product Page](https://www.scaleway.com/en/container-registry/) \n",
        "- [Scaleway Object Storage Product page](https://www.scaleway.com/en/object-storage/)\n",
        "- [Scaleway Object Storage documentation](https://www.scaleway.com/en/docs/object-storage-feature/) \n",
        "- [Scaleway Block Storage Product page](https://www.scaleway.com/en/block-storage/)\n",
        "- [Scaleway Block Storage documentation](https://www.scaleway.com/en/docs/block-storage-overview/)\n",
        "- [Access to the Scaleway Console](https://console.scaleway.com/kapsule/)\n",
        "- [Access to the Kubernetes Console](http://localhost:8001/api/v1/namespaces/kubernetes-dashboard/services/https:kubernetes-dashboard:/proxy/#/login)\n",
        "- [Access to the Kubeflow Dashboard](http://localhost:8080/)\n",
        " "
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "7o8WaNs5gWW4"
      },
      "source": [
        "**Kubeflow Pipeline : Main concepts**\n",
        "\n",
        "- A pipeline is defined in a Python function, which is then compiled in a DSL format for execution.\n",
        "- A pipeline is composed of tasks (components), either defined in :\n",
        "    - a `dsl.ContainerOp()` object, which execute a Docker Container containing the code to run for this task (Packaged in a docker image, this pipeline component is easily shareable and reusable)\n",
        "    - or in a python function that will be converted on the fly into a Docker Container, thanks to the  `kfp.components.func_to_container_op()` API function (Great to prototype pipelines)\n",
        "- Building a pipeline, is as simple as using some task's output as input from other tasks\n",
        "- In order to be able to link the tasks in the pipelines, Kubeflow Pipelines needs to know the type of the input and output parameters of the python functions implementing the tasks\n",
        "- Tasks can be specified to access a volume storage, and/or to use GPU ressources\n",
        "- In order to execute the Docker Container running the pipeline task/components, the Docker images must contain all the python libraries requied by the code\n",
        "    - When converting a python function into a ContainerOp via `kfp.components.func_to_container_op()`, the python imports must be included in the body of this Python function "
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "f-a2V74KgWW5"
      },
      "source": [
        "## Additional Package Installation\n",
        "\n",
        "(You only need to execute the following 2 cells the first time you setup your Jupyter Server environment)"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "VoRa0JxTgWW7"
      },
      "source": [
        "!pip install kfp-server-api=='0.5.0' --user\n",
        "!pip install kfp --upgrade"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "cowZWDSmgWW8"
      },
      "source": [
        "# Need to restart the Jupyter Kernel\n",
        "import os\n",
        "os._exit(00)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "DNo4XL0IgWW9"
      },
      "source": [
        "## Notebook Setup"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "AONyFr2pgWW-"
      },
      "source": [
        "# -------------------------------------\n",
        "#    Notebook configuration 'magic'\n",
        "# -------------------------------------\n",
        "%load_ext autoreload\n",
        "%autoreload 2\n",
        "%matplotlib inline\n",
        "%config InlineBackend.figure_format = 'retina'"
      ],
      "execution_count": 1,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "ncXeuCY2gWW-"
      },
      "source": [
        "# -------------------------------------\n",
        "#     Import Kubeflow Pipelines SDK \n",
        "# -------------------------------------\n",
        "import kfp\n",
        "import kfp.dsl as dsl\n",
        "import kfp.notebook\n",
        "import kfp.components as comp\n",
        "from kfp import compiler\n",
        "from kfp.components import func_to_container_op, InputPath, OutputPath\n",
        "from kubernetes import client as k8s_client\n"
      ],
      "execution_count": 2,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "h66RYTz1gWW_"
      },
      "source": [
        "## Pipeline Component : Download the dataset\n",
        "\n",
        "In this pipeline task, we will:\n",
        "- Create a python function to download the dataset\n",
        "- Have Kubeflow to convert the python function into a ContainerOp (basically this package the python function into a Docker Image, here based on the Tensorflow Docker image)\n",
        "\n",
        "In this example the dataset has been collected from BookCrossing.com, a website dedicated to the practice of \"releasing books into the wild\" - leaving them in public places to be picked up and read by other members of the community.  \n",
        "\n",
        "The downloaded datasets will be stored on the NFS Server (stored on peristed Block Storage volume created during the Kubeflow cluster setup). The Data will be located in the directory `/mnt/nfs/data/datasets/`"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "D3PtF9H_gWXA"
      },
      "source": [
        "def download_dataset(fname: str, origin: str, extract: bool = True,\n",
        "                     cachedir: str = \"./\", cachesubdir: str = 'datasets')-> str:\n",
        "    import tensorflow as tf\n",
        "    import os  \n",
        "    \n",
        "    try:\n",
        "        # Use Keras.utils to download the dataset archive\n",
        "        data_path = tf.keras.utils.get_file(fname, origin,\n",
        "                          extract=extract,\n",
        "                          archive_format='auto',\n",
        "                          cache_dir=cachedir,\n",
        "                          cache_subdir=cachesubdir)\n",
        "\n",
        "        output_dir = os.path.dirname(data_path)\n",
        "        print(\"Path location to the dataset is {}\".format(output_dir))\n",
        "        print(\"{} contains {}\".format(output_dir, os.listdir(output_dir)))\n",
        "        \n",
        "    except ConnectionError:\n",
        "        print('Failed to download the dataset at url {}'.format(origin))\n",
        "        return None\n",
        "    \n",
        "    # ------------------------------\n",
        "    #     Write the Output of the\n",
        "    #   Kubeflow Pipeline Component\n",
        "    # ------------------------------\n",
        "    try:\n",
        "      # This works only inside Docker containers\n",
        "      with open('/output.txt', 'w') as f:\n",
        "        f.write(output_dir)\n",
        "\n",
        "    except PermissionError:\n",
        "        pass\n",
        "    \n",
        "    return output_dir\n",
        "    \n",
        "# -----------------------------------------\n",
        "#        Convert the Python Function\n",
        "#   into a Kubeflow Pipeline ContainerOp\n",
        "# -----------------------------------------     \n",
        "download_op = comp.func_to_container_op(download_dataset,\n",
        "                                        base_image='tensorflow/tensorflow:latest')\n",
        "\n"
      ],
      "execution_count": 3,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "EKPlnYOpgWXB"
      },
      "source": [
        "## Pipeline Component : Prepare the Dataset\n",
        "\n",
        "In this pipeline task, we will: Clean and prepare the dataset. The output of this task will be store on the NFS server in `/mnt/nfs/data/datasets/matrix.pickle`"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "GgBvQHdegWXC"
      },
      "source": [
        "def prepare_dataset(datadir: str)-> str:\n",
        "\n",
        "    import pandas as pd\n",
        "    import time\n",
        "    \n",
        "    print(\"Reading data from\", datadir)\n",
        "    \n",
        "    # Load the books and rating Datasets into a Panda Dataframes\n",
        "    books = pd.read_csv(datadir+'/BX-Books.csv', sep=';', error_bad_lines=False, encoding=\"latin-1\")\n",
        "    books.columns = ['ISBN', 'bookTitle', 'bookAuthor', 'yearOfPublication', 'publisher', 'imageUrlS', 'imageUrlM', 'imageUrlL']\n",
        "\n",
        "    ratings = pd.read_csv(datadir+'/BX-Book-Ratings.csv', sep=';', error_bad_lines=False, encoding=\"latin-1\")\n",
        "    ratings.columns = ['userID', 'ISBN', 'bookRating']\n",
        "    \n",
        "    # Keep only Ratings above 5:\n",
        "    ratings = ratings[ratings.bookRating > 5]\n",
        "\n",
        "    # Drop the columns that we are not going to use\n",
        "    columns = ['yearOfPublication', 'publisher', 'imageUrlS', 'imageUrlM', 'imageUrlL']\n",
        "    books = books.drop(columns, axis=1)\n",
        "    books = books.drop_duplicates(subset='ISBN', keep=\"first\")\n",
        "    books = books.set_index('ISBN', verify_integrity=True)\n",
        "    \n",
        "    \n",
        "    # Keep only those books, that have at least 2 ratings:\n",
        "    ratings_count = ratings.groupby(by='ISBN')['bookRating'].count().reset_index().rename(columns={'bookRating':'ratingCount'})\n",
        "\n",
        "    ratings = pd.merge(ratings, ratings_count, on='ISBN')\n",
        "    ratings = ratings[ratings.ratingCount > 2]\n",
        "    ratings = ratings.drop(['ratingCount'], axis=1)\n",
        "\n",
        "    print(\"Rating shape\", ratings.shape[0])\n",
        "    start = time.time()\n",
        "    matrix = ratings.pivot(index='ISBN', columns='userID', values='bookRating').fillna(0)\n",
        "    end = time.time()\n",
        "    print('Time it took to pivot the ratings table: ', end - start)\n",
        "    \n",
        "    # Save Pandas dataframe\n",
        "    output=datadir+'/matrix.pickle'\n",
        "    matrix.to_pickle(output)\n",
        "    \n",
        "    # ------------------------------\n",
        "    #     Write the Output of the\n",
        "    #   Kubeflow Pipeline Component\n",
        "    # ------------------------------\n",
        "    try:\n",
        "      # This works only inside Docker containers\n",
        "      with open('/output.txt', 'w') as f:\n",
        "        f.write(output)\n",
        "\n",
        "    except PermissionError:\n",
        "        pass\n",
        "    \n",
        "    return output\n",
        "    \n",
        "    \n",
        "# -----------------------------------------\n",
        "#        Convert the Python Function\n",
        "#   into a Kubeflow Pipeline ContainerOp\n",
        "# -----------------------------------------    \n",
        "prepare_op = comp.func_to_container_op(prepare_dataset,\n",
        "                                              base_image='tensorflow/tensorflow:latest',\n",
        "                                              packages_to_install=['pandas'])  "
      ],
      "execution_count": 7,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "7yaz0VDBgWXD"
      },
      "source": [
        "## Pipeline Component :  Fit a model with SciKit-Learn and make one book prediction\n",
        "\n",
        "In this pipeline task, we will fit a ScikitLearn NearestNeighbors model as described in Olga's Petrova blogpost \"[CPU or GPU for your recommendation engine ?](https://blog.scaleway.com/2020/cpu-or-gpu-for-your-recommendation-engine/)\". We will also make a few predictions (using hardcoded input features), and monitor the execution time for inference, here on CPU (Scaleway GP1-M instance)"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "LoZLtzQ4gWXD"
      },
      "source": [
        "def recommender_scikit(picklefile: str)-> str:\n",
        "    import pandas as pd\n",
        "    import time\n",
        "    import os\n",
        "    from joblib import dump\n",
        "    \n",
        "    print(\"Reading processed dataset dataframe pickle from\", picklefile)\n",
        "    \n",
        "    # Reload Processed dataset in a Pandas DataFrame\n",
        "    matrix = pd.read_pickle(picklefile)\n",
        "\n",
        "    from scipy.sparse import csr_matrix\n",
        "    from sklearn.neighbors import NearestNeighbors\n",
        "\n",
        "    # Fit the model\n",
        "    start = time.time()\n",
        "    book_matrix = csr_matrix(matrix.values)\n",
        "    recommender = NearestNeighbors(metric='cosine', algorithm='brute', n_neighbors=10).fit(book_matrix)\n",
        "    print('Time to fit the NearestNeighbors model {}'.format(time.time()-start))\n",
        "    \n",
        "    # Compute 3 Book Recommendations inference and monitor the execution time:\n",
        "    start = time.time()\n",
        "    _, nearestBooks = recommender.kneighbors(matrix.loc['059035342X'].values.reshape(1, -1))\n",
        "    print(\"----------------------------------------\")\n",
        "    print('Time to make a recommendation for ISBN 059035342X using the CSR matrix: {}'.format(time.time()-start))\n",
        "\n",
        "    print(\"----------------------------------------\")\n",
        "    start = time.time()\n",
        "    _, nearestBooks = recommender.kneighbors(matrix.loc['0439064872'].values.reshape(1, -1))\n",
        "    print('Time to make a recommendation for ISBN 0439064872 using the CSR matrix: {}'.format(time.time()-start))\n",
        "    \n",
        "    print(\"----------------------------------------\")\n",
        "    start = time.time()\n",
        "    _, nearestBooks = recommender.kneighbors(matrix.loc['0425189058'].values.reshape(1, -1))\n",
        "    print('Time to make a recommendation for ISBN 0425189058 using the CSR matrix: {}'.format(time.time()-start))\n",
        "\n",
        "\n",
        "    # Save the model\n",
        "    output_dir = os.path.dirname(picklefile)\n",
        "    output=output_dir+'/scikit-nearestneighbors.joblib'\n",
        "    dump(recommender, output) \n",
        "    \n",
        "    # ------------------------------\n",
        "    #     Write the Output of the\n",
        "    #   Kubeflow Pipeline Component\n",
        "    # ------------------------------\n",
        "    try:\n",
        "      # This works only inside Docker containers\n",
        "      with open('/output.txt', 'w') as f:\n",
        "        f.write(output)\n",
        "\n",
        "    except PermissionError:\n",
        "        pass\n",
        "\n",
        "    return output\n",
        "\n",
        "# -----------------------------------------\n",
        "#        Convert the Python Function\n",
        "#   into a Kubeflow Pipeline ContainerOp\n",
        "# -----------------------------------------    \n",
        "recommender_scikit_op = comp.func_to_container_op(recommender_scikit,\n",
        "                                              base_image='tensorflow/tensorflow:latest',\n",
        "                                              packages_to_install=['pandas', 'joblib','scikit-learn'])  "
      ],
      "execution_count": 8,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "t7ipPORbgWXF"
      },
      "source": [
        "## Pipeline Component : Fit a model with Pytorch (GPU) and make one book prediction\n",
        "\n",
        "In this pipeline task, we will fit a NearestNeighbors model, but this time using Pytorch which runs on GPU, as described in Olga's Petrova blogpost \"[CPU or GPU for your recommendation engine ?](https://blog.scaleway.com/2020/cpu-or-gpu-for-your-recommendation-engine/)\". We will also make a few predictions (using hardcoded input features), and monitor the execution time for inference, here on GPU (Scaleway Render-S instance with P100)"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "IGIKvgeIgWXF"
      },
      "source": [
        "def recommender_pytorch(picklefile: str)-> str:\n",
        "\n",
        "    import pandas as pd\n",
        "    import time\n",
        "    import os\n",
        "    import torch\n",
        "    \n",
        "    # Reload Processed dataset in a Pandas DataFrame\n",
        "    matrix = pd.read_pickle(picklefile)\n",
        "\n",
        "    # In PyTorch, you need to explicitely specify when you want an \n",
        "    # operation to be carried out on the GPU. \n",
        "    device = torch.device(\"cuda:0\" if torch.cuda.is_available() else \"cpu\")\n",
        "    print('Running on device: ', device)\n",
        "\n",
        "    # Now we are going to simply append .to(device) to all of our torch \n",
        "    # tensors and modules, e.g.:\n",
        "    cos_sim = torch.nn.CosineSimilarity(dim=1, eps=1e-6).to(device)\n",
        "\n",
        "    # We start by transferring our recommendation matrix to the GPU:\n",
        "    torch_matrix = torch.from_numpy(matrix.values).float().to(device)\n",
        "\n",
        "    # Compute 3 Book Recommendations inference and monitor the execution time:\n",
        "    start = time.time()\n",
        "    \n",
        "    ind = matrix.index.get_loc('059035342X')\n",
        "    HPtensor = torch_matrix[ind,:].reshape(1, -1)\n",
        "\n",
        "    # Now we can compute the cosine similarities:\n",
        "    similarities = cos_sim(HPtensor, torch_matrix)\n",
        "    _, nearestBooks = torch.topk(similarities, k=10)   \n",
        "    print('Time to make a recommendation for ISBN 059035342X using PyTorch: {}'.format(time.time()-start))\n",
        "    \n",
        "    print(\"----------------------------------------\")\n",
        "    start = time.time()\n",
        "    \n",
        "    ind = matrix.index.get_loc('0439064872')\n",
        "    HPtensor = torch_matrix[ind,:].reshape(1, -1)\n",
        "\n",
        "    # Now we can compute the cosine similarities:\n",
        "    similarities = cos_sim(HPtensor, torch_matrix)\n",
        "    _, nearestBooks = torch.topk(similarities, k=10)\n",
        "    print('Time to make a recommendation for ISBN 0439064872 using PyTorch: {}'.format(time.time()-start))\n",
        "    \n",
        "    print(\"----------------------------------------\")\n",
        "    start = time.time()\n",
        "    \n",
        "    ind = matrix.index.get_loc('0425189058')\n",
        "    HPtensor = torch_matrix[ind,:].reshape(1, -1)\n",
        "\n",
        "    # Now we can compute the cosine similarities:\n",
        "    similarities = cos_sim(HPtensor, torch_matrix)\n",
        "    _, nearestBooks = torch.topk(similarities, k=10)\n",
        "    print('Time to make a recommendation for ISBN 0425189058 using PyTorch: {}'.format(time.time()-start))\n",
        "    \n",
        "    # Save the model\n",
        "    output_dir = os.path.dirname(picklefile)\n",
        "    output = output_dir + '/recommender.pt'\n",
        "    torch.save(cos_sim.state_dict(), output)\n",
        "    \n",
        "    # ------------------------------\n",
        "    #     Write the Output of the\n",
        "    #   Kubeflow Pipeline Component\n",
        "    # ------------------------------\n",
        "    try:\n",
        "      # This works only inside Docker containers\n",
        "      with open('/output.txt', 'w') as f:\n",
        "        f.write(output)\n",
        "\n",
        "    except PermissionError:\n",
        "        pass\n",
        "\n",
        "    return output\n",
        "\n",
        "# -----------------------------------------\n",
        "#        Convert the Python Function\n",
        "#   into a Kubeflow Pipeline ContainerOp\n",
        "# -----------------------------------------    \n",
        "recommender_pytorch_op = comp.func_to_container_op(recommender_pytorch,\n",
        "                                             base_image='pytorch/pytorch:latest',\n",
        "                                             packages_to_install=['pandas','scikit-learn'])  "
      ],
      "execution_count": 9,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "vIIX40TugWXG"
      },
      "source": [
        "## Build the Pipeline"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "ZAozDpE8gWXG"
      },
      "source": [
        "@dsl.pipeline(\n",
        "    name=\"Book Recommendation Engine \",\n",
        "    description=\"A Basic example to build a recommendation engine using Kubeflow Pipelines\"\n",
        ")\n",
        "def book_recommender():\n",
        "    \n",
        "    def mount_nfs_helper(container_op):\n",
        "        ''' Helper Function to mount a NFS Volume to the ContainerOp task'''\n",
        "        # NFS PVC details\n",
        "        claim_name='nfs'\n",
        "        name='workdir'\n",
        "        mount_path='/mnt/nfs'\n",
        "\n",
        "        # Add andd Mount the NFS volume to the ContainerOp\n",
        "        nfs_pvc = k8s_client.V1PersistentVolumeClaimVolumeSource(claim_name=claim_name)\n",
        "        container_op.add_volume(k8s_client.V1Volume(name=name,\n",
        "                                              persistent_volume_claim=nfs_pvc))\n",
        "        container_op.add_volume_mount(k8s_client.V1VolumeMount(mount_path=mount_path, name=name))\n",
        "        return container_op\n",
        "    \n",
        "    \n",
        "    \n",
        "    # Pipeline's task 1 : Download dataset\n",
        "    download_task = download_op(fname=\"BX-CSV-Dump.zip\", \n",
        "                                origin=\"http://www2.informatik.uni-freiburg.de/~cziegler/BX/BX-CSV-Dump.zip\",\n",
        "                                cachedir=\"/mnt/nfs/data\")\n",
        "    download_task = mount_nfs_helper(download_task)\n",
        " \n",
        "    # Pipeline's task 2 : Prepare the Dataset\n",
        "    prepare_task = prepare_op(datadir=download_task.output)\n",
        "    prepare_task = mount_nfs_helper(prepare_task)\n",
        "\n",
        "    # Pipeline's task 3 : Train the Scikit-learn NearestNeighbors model\n",
        "    recommender_scikit_task = recommender_scikit_op(picklefile=prepare_task.output)\n",
        "    recommender_scikit_task = mount_nfs_helper(recommender_scikit_task)\n",
        " \n",
        "    # Pipeline's task 3 : Fit the model and Prediction for one isbn with Pytorch on GPU (NearestNeighbors)\n",
        "    recommender_pytorch_task = recommender_pytorch_op(picklefile=prepare_task.output)\n",
        "    recommender_pytorch_task = mount_nfs_helper(recommender_pytorch_task)\n",
        "    recommender_pytorch_task.set_gpu_limit(1)\n",
        "\n",
        "    # Pipeline's task 4 : The goal of this task is to trigger a new GPU node to be spawned in the cluster\n",
        "    # It trains the Scikit-learn NearestNeighbors model on a Render-S \n",
        "    # (slightly better execution time than on GTM-1 because the CPU on the Render-S is a higher end model)\n",
        "    recommender_scikit_task2 = recommender_scikit_op(picklefile=prepare_task.output)\n",
        "    recommender_scikit_task2 = mount_nfs_helper(recommender_scikit_task2)\n",
        "    recommender_scikit_task2.set_gpu_limit(1)"
      ],
      "execution_count": 10,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "xYkYJLttgWXH"
      },
      "source": [
        "## Execute the Pipeline"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "kcFzznIHgWXH"
      },
      "source": [
        "#--------------------------------------------------\n",
        "#              Compile the pipeline \n",
        "#        (composed here of 3 tasks)\n",
        "#--------------------------------------------------\n",
        "PACKAGE_NAME = book_recommender.__name__ + '.yaml'\n",
        "kfp.compiler.Compiler().compile(pipeline_func=book_recommender, \n",
        "                                package_path=PACKAGE_NAME)\n",
        "\n",
        "#--------------------------------------------------\n",
        "#      Create/Reuse an Experiment in Kubeflow\n",
        "#--------------------------------------------------\n",
        "EXPERIMENT_NAME = \"Tests\"\n",
        "client = kfp.Client()\n",
        "try:\n",
        "    experiment = client.get_experiment(experiment_name=EXPERIMENT_NAME)\n",
        "except:\n",
        "    experiment = client.create_experiment(EXPERIMENT_NAME)\n",
        "\n",
        "#-------------------------------------------------- \n",
        "#             Submit a pipeline run\n",
        "#\n",
        "#    => This will create a PVC of 20Gi on \n",
        "#          a Block Storage Volume\n",
        "#--------------------------------------------------\n",
        "RUN_NAME = book_recommender.__name__ + ' run'\n",
        "arguments = {}\n",
        "\n",
        "run_result = client.run_pipeline(experiment_id = experiment.id, \n",
        "                                 job_name = RUN_NAME, \n",
        "                                 pipeline_package_path = PACKAGE_NAME,\n",
        "                                 params = arguments\n",
        "                                )\n"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "kPQ1uvXigWXJ"
      },
      "source": [
        "## Compare some results:\n",
        "\n",
        "**Scikit-Learn**\n",
        "```\n",
        "----------------------------------------\n",
        "Time to make a recommendation for ISBN 059035342X using the CSR matrix: 0.023803234100341797\n",
        "----------------------------------------\n",
        "Time to make a recommendation for ISBN 0439064872 using the CSR matrix: 0.016010761260986328\n",
        "----------------------------------------\n",
        "Time to make a recommendation for ISBN 0425189058 using the CSR matrix: 0.008635520935058594\n",
        "```\n",
        "\n",
        "\n",
        "**Pytorch**\n",
        "```\n",
        "Running on device:  cuda:0\n",
        "Time to make a recommendation for ISBN 059035342X using PyTorch: 0.023251771926879883\n",
        "----------------------------------------\n",
        "Time to make a recommendation for ISBN 0439064872 using PyTorch: 0.00032520294189453125\n",
        "----------------------------------------\n",
        "Time to make a recommendation for ISBN 0425189058 using PyTorch: 0.0002586841583251953\n",
        "```\n",
        "\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "pI_0NegUgWXJ"
      },
      "source": [
        "#### Annex: How to access the data on the PVC ?\n",
        "\n",
        "- Create a `nfs_access.yaml` file (adjust the `claimName`)\n",
        "````\n",
        "apiVersion: v1\n",
        "kind: Pod\n",
        "metadata:\n",
        "  name: nfs-access\n",
        "spec:\n",
        "  containers:\n",
        "  - name: bash\n",
        "    image: bash:latest\n",
        "    command: [\"/bin/sh\", \"-ec\", \"while :; do echo '.'; sleep 5 ; done\"]\n",
        "    volumeMounts:\n",
        "    - mountPath: \"/mnt/nfs\"\n",
        "      name: workdir\n",
        "  volumes:\n",
        "  - name: workdir\n",
        "    persistentVolumeClaim:\n",
        "      claimName: nfs\n",
        "```\n",
        "- Create a Pod from this specifications \n",
        "```\n",
        "kubectl apply -f nfs_access.yaml -n kubeflow\n",
        "```\n",
        "- Connect with a shell to this pods (note: there is no prompt on the command line:\n",
        "```\n",
        "kubectl exec -t -i -n kubeflow nfs-access -- /bin/sh\n",
        "# you can explore the /mnt/nfs/data directory from here\n",
        "```\n",
        "- In a similar manner, you can use the kubectl cp command to copy data from/to the PVC\n"
      ]
    }
  ]
}