notebooks/mnist_vanilla.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "!pip install matplotlib numpy scikit-learn tensorflow boto3 botocore"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from IPython.core.display import HTML\n",
    "\n",
    "HTML(\"<script>Jupyter.notebook.kernel.restart()</script>\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Download Datasets"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from os import environ, path\n",
    "from urllib.parse import urlparse\n",
    "from pathlib import Path\n",
    "from typing import List\n",
    "import os\n",
    "\n",
    "import boto3\n",
    "\n",
    "\n",
    "def get_s3_client():\n",
    "    # allow s3 connection without creds\n",
    "    if \"AWS_ACCESS_KEY_ID\" not in environ:\n",
    "        from botocore import UNSIGNED\n",
    "        from botocore.client import Config\n",
    "\n",
    "        return boto3.client(\"s3\", config=Config(signature_version=UNSIGNED))\n",
    "    return boto3.client(\"s3\")\n",
    "\n",
    "\n",
    "def download_s3(bucket, key, outdir):\n",
    "    s3 = get_s3_client()\n",
    "    s3_object = s3.get_object(Bucket=bucket, Key=key)\n",
    "    stream = s3_object[\"Body\"]\n",
    "    outfile = path.join(outdir, key)\n",
    "    filepath = path.abspath(outfile)\n",
    "    parent_dir = path.dirname(outfile)\n",
    "    Path(parent_dir).mkdir(parents=True, exist_ok=True)\n",
    "    with open(outfile, \"wb+\") as f:\n",
    "        f.write(stream.read())\n",
    "    print(f\"file saved to: {outfile}\")\n",
    "\n",
    "\n",
    "def parse_s3_url(url):\n",
    "    print(f\"downloading: {url}\")\n",
    "    u = urlparse(url)\n",
    "    bucket = u.netloc.split(\".\")[0]\n",
    "    key = u.path.strip(\"/\")\n",
    "    return bucket, key\n",
    "\n",
    "\n",
    "def download_s3_dir(data_urls: List[str], data_dir: str):\n",
    "    \"\"\"Download objects from S3\"\"\"\n",
    "\n",
    "    if not os.path.exists(data_dir):\n",
    "        os.makedirs(data_dir)\n",
    "\n",
    "    for data_url in data_urls:\n",
    "        bucket, key = parse_s3_url(data_url)\n",
    "        download_s3(bucket, key, data_dir)\n",
    "\n",
    "\n",
    "def download_data(data_dir: str, data_urls: str):\n",
    "    # data_urls must be type string because kubeflow has no registered serializers for type \"typing.List[str]\"\n",
    "    download_s3_dir(data_urls.split(\",\"), data_dir)\n",
    "    print(\"downloads complete\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "igI1dvp7SWWV"
   },
   "source": [
    "## Train Model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from os import path\n",
    "\n",
    "import tensorflow as tf\n",
    "from tensorflow import keras\n",
    "\n",
    "\n",
    "def gen_log_dirname(log_dir) -> str:\n",
    "    return path.join(log_dir, \"tensorboard\", \"fit\")\n",
    "\n",
    "\n",
    "def load_mnist(filepath, kind, normalize=True):\n",
    "    import gzip\n",
    "    import numpy as np\n",
    "\n",
    "    \"\"\"Load MNIST data from `filepath`\"\"\"\n",
    "    labels_path = path.join(filepath, f\"{kind}-labels-idx1-ubyte.gz\")\n",
    "    images_path = path.join(filepath, f\"{kind}-images-idx3-ubyte.gz\")\n",
    "\n",
    "    with gzip.open(labels_path, \"rb\") as lbpath:\n",
    "        labels = np.frombuffer(lbpath.read(), dtype=np.uint8, offset=8)\n",
    "\n",
    "    with gzip.open(images_path, \"rb\") as imgpath:\n",
    "        images = np.frombuffer(imgpath.read(), dtype=np.uint8, offset=16).reshape(\n",
    "            len(labels), 28, 28\n",
    "        )\n",
    "\n",
    "    # normalize by dividing each pixel value by 255.0. This places the pixel value within the range 0 and 1.\n",
    "    if normalize:\n",
    "        images = images / 255.0\n",
    "\n",
    "    return images, labels\n",
    "\n",
    "\n",
    "def learning_rate(batch_size):\n",
    "    # gradually reduce the learning rate during training\n",
    "    return keras.optimizers.schedules.InverseTimeDecay(\n",
    "        0.001, decay_steps=batch_size * 1000, decay_rate=1, staircase=False\n",
    "    )\n",
    "\n",
    "\n",
    "def create_model(batch_size):\n",
    "    model = keras.Sequential(\n",
    "        [\n",
    "            keras.layers.Flatten(input_shape=(28, 28)),\n",
    "            keras.layers.Dense(128, activation=\"relu\"),\n",
    "            keras.layers.Dropout(0.5),\n",
    "            keras.layers.Dense(128, activation=\"relu\"),\n",
    "            keras.layers.Dense(10),\n",
    "        ]\n",
    "    )\n",
    "    model.compile(\n",
    "        optimizer=keras.optimizers.Adam(\n",
    "            # learning_rate=1e-3,\n",
    "            learning_rate=learning_rate(batch_size),\n",
    "        ),\n",
    "        # https://keras.io/api/metrics/probabilistic_metrics/#sparsecategoricalcrossentropy-class\n",
    "        loss=keras.losses.SparseCategoricalCrossentropy(from_logits=True),\n",
    "        metrics=[\"accuracy\"],\n",
    "    )\n",
    "    model.summary()\n",
    "    return model\n",
    "\n",
    "\n",
    "def train_model(\n",
    "    model_dir: str,\n",
    "    data_dir: str,\n",
    "    log_dir: str,\n",
    "    epochs: int = 5,\n",
    ") -> str:  # noqa: F821\n",
    "    \"\"\"Trains a model and saves to model dir and returns path to tensorboard logs.\"\"\"\n",
    "\n",
    "    train_images, train_labels = load_mnist(data_dir, kind=\"train\", normalize=True)\n",
    "    test_images, test_labels = load_mnist(data_dir, kind=\"t10k\", normalize=True)\n",
    "\n",
    "    tensorboard_log_dir = gen_log_dirname(log_dir)\n",
    "\n",
    "    batch_size = len(train_images)\n",
    "    model = create_model(batch_size)\n",
    "\n",
    "    model.fit(\n",
    "        x=train_images,\n",
    "        y=train_labels,\n",
    "        epochs=epochs,\n",
    "        shuffle=True,\n",
    "        # tensorboard args\n",
    "        validation_data=(test_images, test_labels),\n",
    "        callbacks=[\n",
    "            tf.keras.callbacks.TensorBoard(\n",
    "                log_dir=tensorboard_log_dir, histogram_freq=1\n",
    "            )\n",
    "        ],\n",
    "    )\n",
    "\n",
    "    model.save(model_dir, include_optimizer=True)\n",
    "    return tensorboard_log_dir"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Evaluate Model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "_pxGKEae6Eo_"
   },
   "outputs": [],
   "source": [
    "import json\n",
    "\n",
    "import tensorflow as tf\n",
    "from tensorflow import keras\n",
    "from tensorflow.python.lib.io import file_io\n",
    "import numpy as np\n",
    "from sklearn.metrics import confusion_matrix, accuracy_score\n",
    "\n",
    "from typing import NamedTuple\n",
    "from collections import namedtuple\n",
    "\n",
    "\n",
    "def write_cm_to_csv(cm, class_labels, cm_path):\n",
    "    data = []\n",
    "    for target_index, target_row in enumerate(cm):\n",
    "        for predicted_index, count in enumerate(target_row):\n",
    "            data.append(\n",
    "                (class_labels[target_index], class_labels[predicted_index], count)\n",
    "            )\n",
    "\n",
    "    df_cm = pd.DataFrame(data, columns=[\"target\", \"predicted\", \"count\"])\n",
    "    with file_io.FileIO(cm_path, \"w\") as f:\n",
    "        df_cm.to_csv(\n",
    "            f, columns=[\"target\", \"predicted\", \"count\"], header=False, index=False\n",
    "        )\n",
    "\n",
    "\n",
    "def predict(model, test_images):\n",
    "    # Define a Softmax layer to define outputs as probabilities\n",
    "    probability_model = tf.keras.Sequential([model, tf.keras.layers.Softmax()])\n",
    "    predictions = probability_model.predict(test_images)\n",
    "    return np.ravel(np.matrix(predictions).argmax(1))\n",
    "\n",
    "\n",
    "def evaluate_model(\n",
    "    metrics_path: str,\n",
    "    data_dir: str,\n",
    "    model_dir: str,\n",
    ") -> NamedTuple(\n",
    "    \"output\",\n",
    "    # https://www.kubeflow.org/docs/pipelines/sdk/pipelines-metrics/\n",
    "    # The output name must be MLPipeline Metrics or MLPipeline_Metrics (case does not matter).\n",
    "    [(\"mlpipeline_ui_metadata\", \"UI_metadata\"), (\"mlpipeline_metrics\", \"Metrics\")],\n",
    "):\n",
    "\n",
    "    test_images, test_labels = load_mnist(data_dir, kind=\"t10k\", normalize=False)\n",
    "    model = keras.models.load_model(model_dir)\n",
    "\n",
    "    loss, accuracy = model.evaluate(test_images, test_labels)\n",
    "    metrics = {\n",
    "        \"metrics\": [\n",
    "            {\"name\": \"loss\", \"numberValue\": str(loss), \"format\": \"PERCENTAGE\"},\n",
    "            {\"name\": \"accuracy\", \"numberValue\": str(accuracy), \"format\": \"PERCENTAGE\"},\n",
    "        ]\n",
    "    }\n",
    "    with open(metrics_path, \"w+\") as f:\n",
    "        json.dump(metrics, f)\n",
    "\n",
    "    print_output = namedtuple(\n",
    "        # \"pipeline_metrics\" is hardcoded value that could be anything\n",
    "        \"output\",\n",
    "        [\"pipeline_metrics\"],\n",
    "    )\n",
    "    return print_output(json.dumps(metrics))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Export Model to S3"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from os import environ, path\n",
    "import boto3\n",
    "from botocore.exceptions import ClientError\n",
    "\n",
    "\n",
    "def get_s3_client():\n",
    "    # allow s3 connection without creds\n",
    "    if \"AWS_ACCESS_KEY_ID\" not in environ:\n",
    "        from botocore import UNSIGNED\n",
    "        from botocore.client import Config\n",
    "\n",
    "        return boto3.client(\"s3\", config=Config(signature_version=UNSIGNED))\n",
    "    return boto3.client(\"s3\")\n",
    "\n",
    "\n",
    "def upload_file(s3_client, file_name, bucket, object_name=None):\n",
    "    \"\"\"Upload a file to an S3 bucket\n",
    "\n",
    "    :param file_name: File to upload\n",
    "    :param bucket: Bucket to upload to\n",
    "    :param object_name: S3 object name. If not specified then file_name is used\n",
    "    :return: True if file was uploaded, else False\n",
    "    \"\"\"\n",
    "    # If S3 object_name was not specified, use file_name\n",
    "    if object_name is None:\n",
    "        object_name = file_name\n",
    "\n",
    "    return s3_client.upload_file(file_name, bucket, object_name)\n",
    "\n",
    "\n",
    "def bucket_exists(s3_client, bucket_name):\n",
    "    exists = True\n",
    "    try:\n",
    "        s3_client.head_bucket(Bucket=bucket_name)\n",
    "    except botocore.exceptions.ClientError as e:\n",
    "        # If a client error is thrown, then check that it was a 404 error.\n",
    "        # If it was a 404 error, then the bucket does not exist.\n",
    "        error_code = e.response[\"Error\"][\"Code\"]\n",
    "        if error_code == \"404\":\n",
    "            exists = False\n",
    "    return exists\n",
    "\n",
    "\n",
    "def s3_upload_dir(\n",
    "    src_dir: str,\n",
    "    bucket_name: str,\n",
    "    bucket_dir: str,\n",
    "    # model_dir: InputPath(str), bucket_name: str, bucket_dir: str,\n",
    "):\n",
    "    s3_client = get_s3_client()\n",
    "\n",
    "    if not bucket_exists(s3_client, bucket_name):\n",
    "        raise Exception(f\"Bucket: {bucket_name} does not exist\")\n",
    "\n",
    "    for root, dirs, files in os.walk(model_dir):\n",
    "        for name in files:\n",
    "            local_path = os.path.join(root, name)\n",
    "            upload_file(\n",
    "                s3_client,\n",
    "                local_path,\n",
    "                bucket_name,\n",
    "                f\"{bucket_dir}/{os.path.relpath(local_path, model_dir)}\",\n",
    "            )\n",
    "\n",
    "    response = s3_client.list_objects(Bucket=bucket_name)\n",
    "    print(f\"All objects in {bucket_name}:\")\n",
    "\n",
    "    for file in response[\"Contents\"]:\n",
    "        print(f\"{bucket_name}/{file['Key']}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Run entire pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# !pip install python-dotenv\n",
    "# from dotenv import load_dotenv\n",
    "\n",
    "# # %load_ext dotenv\n",
    "# # !dotenv --help\n",
    "# # !dotenv load -f '../.env' set\n",
    "# load_dotenv(dotenv_path=\"../.env\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "model_version = 1\n",
    "\n",
    "# download_data component\n",
    "data_dir = os.path.join(\"tmp\", \"datasets\")\n",
    "mnist_data_s3_urls = [\n",
    "    \"https://fashion-mnist.s3-website.eu-central-1.amazonaws.com/train-images-idx3-ubyte.gz\",\n",
    "    \"https://fashion-mnist.s3-website.eu-central-1.amazonaws.com/train-labels-idx1-ubyte.gz\",\n",
    "    \"https://fashion-mnist.s3-website.eu-central-1.amazonaws.com/t10k-images-idx3-ubyte.gz\",\n",
    "    \"https://fashion-mnist.s3-website.eu-central-1.amazonaws.com/t10k-labels-idx1-ubyte.gz\",\n",
    "]\n",
    "\n",
    "# train_model component\n",
    "model_dir = os.path.join(\"tmp\", \"model\", str(model_version))\n",
    "log_dir = os.path.join(\"tmp\", \"logs\")\n",
    "metadata_file = os.path.join(\"tmp\", \"metadata.json\")\n",
    "\n",
    "# evaluate_model component\n",
    "metrics_path = os.path.join(\"tmp\", \"metrics.json\")\n",
    "\n",
    "# export model component\n",
    "bucket_name = \"kfaas-demo-data-sandbox\"\n",
    "# bucket_name = \"kfaas-demo-data-prod\"\n",
    "\n",
    "client = \"demo\"\n",
    "model_name = \"fashion_model\"\n",
    "bucket_dir_tensorboard = f\"{client}/tensorboard/{model_name}/{model_version}\"\n",
    "bucket_dir_model = f\"{client}/{model_name}/{model_version}\"\n",
    "\n",
    "# run pipeline\n",
    "download_data(data_dir, \",\".join(mnist_data_s3_urls))\n",
    "tensorboard_log_dir = train_model(\n",
    "    model_dir=model_dir,\n",
    "    data_dir=data_dir,\n",
    "    log_dir=log_dir,\n",
    "    epochs=10,\n",
    ")\n",
    "print(f\"tensorboard_log_dir={tensorboard_log_dir}\")\n",
    "evaluate_model(metrics_path, data_dir, model_dir)\n",
    "!saved_model_cli show --dir \"tmp/model/1\" --all\n",
    "# s3_upload_dir(model_dir, bucket_name, bucket_dir_model)\n",
    "# s3_upload_dir(tensorboard_log_dir, bucket_name, bucket_dir_model)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Run Tensorboard on model training logs\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!python -c \"import sklearn; sklearn.show_versions()\"\n",
    "# Load the TensorBoard notebook extension\n",
    "%reload_ext tensorboard\n",
    "%tensorboard --logdir tmp/logs/tensorboard/fit\n",
    "!echo \"go to http://localhost:6006/\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from tensorboard import notebook\n",
    "\n",
    "notebook.list()  # View open TensorBoard instances"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Tensorflow Serving\n",
    "- Do not include model version in source path (ie NOT `${PWD}/tmp/model/1`)\n",
    "- Do not use environment variables, must hardcode model_name\n",
    "### Run with docker\n",
    "```console\n",
    "cd notebooks\n",
    "docker run -p 8501:8501 \\\n",
    "--mount type=bind,source=$PWD/tmp/model,target=/models/fashion_model \\\n",
    "-e MODEL_NAME=fashion_model -t tensorflow/serving\n",
    "```\n",
    "\n",
    "### Query\n",
    "```shell\n",
    "http get http://0.0.0.0:8501/v1/models/fashion_model\n",
    "http get http://0.0.0.0:8501/v1/models/fashion_model/metatdata\n",
    "```\n",
    "\n",
    "### Install TFServing with APT\n",
    "```bash\n",
    "sudo bash\n",
    "echo \"deb http://storage.googleapis.com/tensorflow-serving-apt stable tensorflow-model-server tensorflow-model-server-universal\" | tee /etc/apt/sources.list.d/tensorflow-serving.list && \\\n",
    "curl https://storage.googleapis.com/tensorflow-serving-apt/tensorflow-serving.release.pub.gpg | apt-key add -\n",
    "apt update\n",
    "apt-get install tensorflow-model-server\n",
    "```\n",
    "\n",
    "### Run with APT installed\n",
    "```bash\n",
    "export MODEL_NAME=fashion_model\n",
    "tensorflow_model_server \\\n",
    "  --rest_api_port=8501 \\\n",
    "  --model_name=fashion_model \\\n",
    "  --model_base_path=\"$PWD/tmp/model\"\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!pip install requests"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Query Tensorflow Server API"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class_labels = [\n",
    "    \"T-shirt/top\",\n",
    "    \"Trouser\",\n",
    "    \"Pullover\",\n",
    "    \"Dress\",\n",
    "    \"Coat\",\n",
    "    \"Sandal\",\n",
    "    \"Shirt\",\n",
    "    \"Sneaker\",\n",
    "    \"Bag\",\n",
    "    \"Ankle boot\",\n",
    "]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import matplotlib.pyplot as plt\n",
    "import requests\n",
    "import json\n",
    "\n",
    "\n",
    "def show(idx, title):\n",
    "    plt.figure()\n",
    "    plt.imshow(test_images[idx].reshape(28, 28))\n",
    "    plt.axis(\"off\")\n",
    "    plt.title(\"\\n\\n{}\".format(title), fontdict={\"size\": 16})\n",
    "\n",
    "\n",
    "def query_api(model_name, data, version=1, port=8501):\n",
    "    headers = {\"content-type\": \"application/json\"}\n",
    "    r = requests.post(\n",
    "        f\"http://0.0.0.0:{port}/v1/models/{model_name}/versions/{version}:predict\",\n",
    "        # f\"http://0.0.0.0:{port}/v1/models/{model_name}:predict\",\n",
    "        data=json.dumps(data),\n",
    "        headers=headers,\n",
    "    )\n",
    "    res = r.json()\n",
    "    if \"error\" in res:\n",
    "        raise Exception(res[\"error\"])\n",
    "\n",
    "    # for i in range(0,len(json.loads(data)[\"instances\"])):\n",
    "    for i, pred in enumerate(res[\"predictions\"]):\n",
    "        show(\n",
    "            i,\n",
    "            \"The model thought this was a {} (class {}), and it was actually a {} (class {})\".format(\n",
    "                class_labels[np.argmax(pred)],\n",
    "                np.argmax(pred),\n",
    "                class_labels[data[\"labels\"][i]],\n",
    "                data[\"labels\"][i],\n",
    "            ),\n",
    "        )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import json\n",
    "\n",
    "test_images, test_labels = load_mnist(data_dir, kind=\"t10k\", normalize=True)\n",
    "data = {\n",
    "    \"signature_name\": \"serving_default\",\n",
    "    \"instances\": test_images[0:3].tolist(),\n",
    "    \"labels\": test_labels[0:3].tolist(),\n",
    "}\n",
    "query_api(model_name, data, model_version, port=8501)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import random\n",
    "\n",
    "test_images, test_labels = load_mnist(data_dir, kind=\"t10k\", normalize=True)\n",
    "rando = random.randint(0, len(test_images) - 1)\n",
    "show(rando, \"An Example Image: {}\".format(class_labels[test_labels[rando]]))"
   ]
  }
 ],
 "metadata": {
  "colab": {
   "collapsed_sections": [],
   "name": "KF Fashion MNIST",
   "provenance": []
  },
  "kernelspec": {
   "display_name": "Python 3.8.6 64-bit ('env')",
   "metadata": {
    "interpreter": {
     "hash": "87baad19644053012dbfcc6c6f15b83806c5b811f1a83788faf1a84416645ee7"
    }
   },
   "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.9.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}