digits_recognizer_pipeline.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "5da13c87-aed3-4c3a-bac8-1807ded41b79",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<a href=\"/pipeline/#/experiments/details/e5076f7c-8815-4eb9-81e2-223e31a65748\" target=\"_blank\" >Experiment details</a>."
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/html": [
       "<a href=\"/pipeline/#/runs/details/ff09ee4c-feac-4d47-b2eb-d177b7696c00\" target=\"_blank\" >Run details</a>."
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# \n",
    "# Creating a ML pipeline with the MNIST digits dataset\n",
    "# KFPv1 Example with lightweight Python components only\n",
    "#\n",
    "\n",
    "import kfp\n",
    "from kfp import dsl\n",
    "import kfp.components as components\n",
    "\n",
    "from typing import NamedTuple\n",
    "def get_data_batch() -> NamedTuple('Outputs', [('datapoints_training', float),('datapoints_test', float),('dataset_version', str)]):\n",
    "    \"\"\"\n",
    "    Function to get dataset and load it to minio bucket\n",
    "    \"\"\"\n",
    "    print(\"getting data\")\n",
    "    from tensorflow import keras\n",
    "    from minio import Minio\n",
    "    import numpy as np\n",
    "    import json\n",
    "\n",
    "    minio_client = Minio(\n",
    "        \"100.65.11.110:9000\",\n",
    "        access_key=\"minio\",\n",
    "        secret_key=\"minio123\",\n",
    "        secure=False\n",
    "    )\n",
    "    minio_bucket = \"mlpipeline\"\n",
    "    \n",
    "    minio_client.fget_object(minio_bucket,\"mnist.npz\",\"/tmp/mnist.npz\")\n",
    "    \n",
    "    def load_data():\n",
    "        with np.load(\"/tmp/mnist.npz\", allow_pickle=True) as f:\n",
    "            x_train, y_train = f[\"x_train\"], f[\"y_train\"]\n",
    "            x_test, y_test = f[\"x_test\"], f[\"y_test\"]\n",
    "\n",
    "        return (x_train, y_train), (x_test, y_test)\n",
    "    \n",
    "    # Get MNIST data directly from library\n",
    "    (x_train, y_train), (x_test, y_test) = load_data()\n",
    "\n",
    "    # save to numpy file, store in Minio\n",
    "    np.save(\"/tmp/x_train.npy\",x_train)\n",
    "    minio_client.fput_object(minio_bucket,\"x_train\",\"/tmp/x_train.npy\")\n",
    "\n",
    "    np.save(\"/tmp/y_train.npy\",y_train)\n",
    "    minio_client.fput_object(minio_bucket,\"y_train\",\"/tmp/y_train.npy\")\n",
    "\n",
    "    np.save(\"/tmp/x_test.npy\",x_test)\n",
    "    minio_client.fput_object(minio_bucket,\"x_test\",\"/tmp/x_test.npy\")\n",
    "\n",
    "    np.save(\"/tmp/y_test.npy\",y_test)\n",
    "    minio_client.fput_object(minio_bucket,\"y_test\",\"/tmp/y_test.npy\")\n",
    "    \n",
    "    dataset_version = \"1.0\"\n",
    "    \n",
    "    print(f\"x_train shape: {x_train.shape}\")\n",
    "    print(f\"y_train shape: {y_train.shape}\")\n",
    "\n",
    "    print(f\"x_test shape: {x_test.shape}\")\n",
    "    print(f\"y_test shape: {y_test.shape}\")\n",
    "    \n",
    "    from collections import namedtuple\n",
    "    divmod_output = namedtuple('Outputs', ['datapoints_training', 'datapoints_test', 'dataset_version'])\n",
    "    return [float(x_train.shape[0]),float(x_test.shape[0]),dataset_version]\n",
    "    \n",
    "def get_latest_data():\n",
    "    \"\"\"\n",
    "    Dummy functions for showcasing\n",
    "    \"\"\"\n",
    "    print(\"Adding latest data\")\n",
    "    \n",
    "    \n",
    "def reshape_data():\n",
    "    \"\"\"\n",
    "    Reshape the data for model building\n",
    "    \"\"\"\n",
    "    print(\"reshaping data\")\n",
    "    \n",
    "    from minio import Minio\n",
    "    import numpy as np\n",
    "\n",
    "    minio_client = Minio(\n",
    "        \"100.65.11.110:9000\",\n",
    "        access_key=\"minio\",\n",
    "        secret_key=\"minio123\",\n",
    "        secure=False\n",
    "    )\n",
    "    minio_bucket = \"mlpipeline\"\n",
    "    \n",
    "    # load data from minio\n",
    "    minio_client.fget_object(minio_bucket,\"x_train\",\"/tmp/x_train.npy\")\n",
    "    x_train = np.load(\"/tmp/x_train.npy\")\n",
    "    \n",
    "    minio_client.fget_object(minio_bucket,\"x_test\",\"/tmp/x_test.npy\")\n",
    "    x_test = np.load(\"/tmp/x_test.npy\")\n",
    "    \n",
    "    # reshaping the data\n",
    "    # reshaping pixels in a 28x28px image with greyscale, canal = 1. This is needed for the Keras API\n",
    "    x_train = x_train.reshape(-1,28,28,1)\n",
    "    x_test = x_test.reshape(-1,28,28,1)\n",
    "\n",
    "    # normalizing the data\n",
    "    # each pixel has a value between 0-255. Here we divide by 255, to get values from 0-1\n",
    "    x_train = x_train / 255\n",
    "    x_test = x_test / 255\n",
    "    \n",
    "    # save data from minio\n",
    "    np.save(\"/tmp/x_train.npy\",x_train)\n",
    "    minio_client.fput_object(minio_bucket,\"x_train\",\"/tmp/x_train.npy\")\n",
    "    \n",
    "    np.save(\"/tmp/x_test.npy\",x_test)\n",
    "    minio_client.fput_object(minio_bucket,\"x_test\",\"/tmp/x_test.npy\")\n",
    "\n",
    "def model_building(\n",
    "    no_epochs:int = 1,\n",
    "    optimizer: str = \"adam\"\n",
    ") -> NamedTuple('Output', [('mlpipeline_ui_metadata', 'UI_metadata'),('mlpipeline_metrics', 'Metrics')]):\n",
    "    \"\"\"\n",
    "    Build the model with Keras API\n",
    "    Export model parameters\n",
    "    \"\"\"\n",
    "    from tensorflow import keras\n",
    "    import tensorflow as tf\n",
    "    from minio import Minio\n",
    "    import numpy as np\n",
    "    import pandas as pd\n",
    "    import json\n",
    "    \n",
    "    minio_client = Minio(\n",
    "        \"100.65.11.110:9000\",\n",
    "        access_key=\"minio\",\n",
    "        secret_key=\"minio123\",\n",
    "        secure=False\n",
    "    )\n",
    "    minio_bucket = \"mlpipeline\"\n",
    "    \n",
    "    model = keras.models.Sequential()\n",
    "    model.add(keras.layers.Conv2D(64, (3, 3), activation='relu', input_shape=(28,28,1)))\n",
    "    model.add(keras.layers.MaxPool2D(2, 2))\n",
    "\n",
    "    model.add(keras.layers.Flatten())\n",
    "    model.add(keras.layers.Dense(64, activation='relu'))\n",
    "\n",
    "    model.add(keras.layers.Dense(32, activation='relu'))\n",
    "\n",
    "    model.add(keras.layers.Dense(10, activation='softmax')) #output are 10 classes, numbers from 0-9\n",
    "\n",
    "    #show model summary - how it looks\n",
    "    stringlist = []\n",
    "    model.summary(print_fn=lambda x: stringlist.append(x))\n",
    "    metric_model_summary = \"\\n\".join(stringlist)\n",
    "    \n",
    "    #compile the model - we want to have a binary outcome\n",
    "    model.compile(optimizer=optimizer,\n",
    "              loss=\"sparse_categorical_crossentropy\",\n",
    "              metrics=['accuracy'])\n",
    "    \n",
    "    minio_client.fget_object(minio_bucket,\"x_train\",\"/tmp/x_train.npy\")\n",
    "    x_train = np.load(\"/tmp/x_train.npy\")\n",
    "    \n",
    "    minio_client.fget_object(minio_bucket,\"y_train\",\"/tmp/y_train.npy\")\n",
    "    y_train = np.load(\"/tmp/y_train.npy\")\n",
    "    \n",
    "    #fit the model and return the history while training\n",
    "    history = model.fit(\n",
    "      x=x_train,\n",
    "      y=y_train,\n",
    "      epochs=no_epochs,\n",
    "      batch_size=20,\n",
    "    )\n",
    "    \n",
    "    minio_client.fget_object(minio_bucket,\"x_test\",\"/tmp/x_test.npy\")\n",
    "    x_test = np.load(\"/tmp/x_test.npy\")\n",
    "    \n",
    "    minio_client.fget_object(minio_bucket,\"y_test\",\"/tmp/y_test.npy\")\n",
    "    y_test = np.load(\"/tmp/y_test.npy\")\n",
    "    \n",
    "\n",
    "    # Test the model against the test dataset\n",
    "    # Returns the loss value & metrics values for the model in test mode.\n",
    "    model_loss, model_accuracy = model.evaluate(x=x_test,y=y_test)\n",
    "    \n",
    "    # Confusion Matrix\n",
    "\n",
    "    # Generates output predictions for the input samples.\n",
    "    test_predictions = model.predict(x=x_test)\n",
    "\n",
    "    # Returns the indices of the maximum values along an axis.\n",
    "    test_predictions = np.argmax(test_predictions,axis=1) # the prediction outputs 10 values, we take the index number of the highest value, which is the prediction of the model\n",
    "\n",
    "    # generate confusion matrix\n",
    "    confusion_matrix = tf.math.confusion_matrix(labels=y_test,predictions=test_predictions)\n",
    "    confusion_matrix = confusion_matrix.numpy()\n",
    "    vocab = list(np.unique(y_test))\n",
    "    data = []\n",
    "    for target_index, target_row in enumerate(confusion_matrix):\n",
    "        for predicted_index, count in enumerate(target_row):\n",
    "            data.append((vocab[target_index], vocab[predicted_index], count))\n",
    "\n",
    "    df_cm = pd.DataFrame(data, columns=['target', 'predicted', 'count'])\n",
    "    cm_csv = df_cm.to_csv(header=False, index=False)\n",
    "    \n",
    "    metadata = {\n",
    "        \"outputs\": [\n",
    "            {\n",
    "                \"type\": \"confusion_matrix\",\n",
    "                \"format\": \"csv\",\n",
    "                \"schema\": [\n",
    "                    {'name': 'target', 'type': 'CATEGORY'},\n",
    "                    {'name': 'predicted', 'type': 'CATEGORY'},\n",
    "                    {'name': 'count', 'type': 'NUMBER'},\n",
    "                  ],\n",
    "                \"target_col\" : \"actual\",\n",
    "                \"predicted_col\" : \"predicted\",\n",
    "                \"source\": cm_csv,\n",
    "                \"storage\": \"inline\",\n",
    "                \"labels\": [0,1,2,3,4,5,6,7,8,9]\n",
    "            },\n",
    "            {\n",
    "                'storage': 'inline',\n",
    "                'source': '''# Model Overview\n",
    "## Model Summary\n",
    "\n",
    "```\n",
    "{}\n",
    "```\n",
    "\n",
    "## Model Performance\n",
    "\n",
    "**Accuracy**: {}\n",
    "**Loss**: {}\n",
    "\n",
    "'''.format(metric_model_summary,model_accuracy,model_loss),\n",
    "                'type': 'markdown',\n",
    "            }\n",
    "        ]\n",
    "    }\n",
    "    \n",
    "    metrics = {\n",
    "      'metrics': [{\n",
    "          'name': 'model_accuracy',\n",
    "          'numberValue':  float(model_accuracy),\n",
    "          'format' : \"PERCENTAGE\"\n",
    "        },{\n",
    "          'name': 'model_loss',\n",
    "          'numberValue':  float(model_loss),\n",
    "          'format' : \"PERCENTAGE\"\n",
    "        }]}\n",
    "    \n",
    "    ### Save model to minIO\n",
    "    \n",
    "    keras.models.save_model(model,\"/tmp/detect-digits\")\n",
    "    \n",
    "    from minio import Minio\n",
    "    import os\n",
    "\n",
    "    minio_client = Minio(\n",
    "            \"100.65.11.110:9000\",\n",
    "            access_key=\"minio\",\n",
    "            secret_key=\"minio123\",\n",
    "            secure=False\n",
    "        )\n",
    "    minio_bucket = \"mlpipeline\"\n",
    "\n",
    "\n",
    "    import glob\n",
    "\n",
    "    def upload_local_directory_to_minio(local_path, bucket_name, minio_path):\n",
    "        assert os.path.isdir(local_path)\n",
    "\n",
    "        for local_file in glob.glob(local_path + '/**'):\n",
    "            local_file = local_file.replace(os.sep, \"/\") # Replace \\ with / on Windows\n",
    "            if not os.path.isfile(local_file):\n",
    "                upload_local_directory_to_minio(\n",
    "                    local_file, bucket_name, minio_path + \"/\" + os.path.basename(local_file))\n",
    "            else:\n",
    "                remote_path = os.path.join(\n",
    "                    minio_path, local_file[1 + len(local_path):])\n",
    "                remote_path = remote_path.replace(\n",
    "                    os.sep, \"/\")  # Replace \\ with / on Windows\n",
    "                minio_client.fput_object(bucket_name, remote_path, local_file)\n",
    "\n",
    "    upload_local_directory_to_minio(\"/tmp/detect-digits\",minio_bucket,\"models/detect-digits/1/\") # 1 for version 1\n",
    "    \n",
    "    print(\"Saved model to minIO\")\n",
    "    \n",
    "    from collections import namedtuple\n",
    "    output = namedtuple('output', ['mlpipeline_ui_metadata', 'mlpipeline_metrics'])\n",
    "    return output(json.dumps(metadata),json.dumps(metrics))\n",
    "\n",
    "def model_serving():\n",
    "    \"\"\"\n",
    "    Create kserve instance\n",
    "    \"\"\"\n",
    "    from kubernetes import client \n",
    "    from kserve import KServeClient\n",
    "    from kserve import constants\n",
    "    from kserve import utils\n",
    "    from kserve import V1beta1InferenceService\n",
    "    from kserve import V1beta1InferenceServiceSpec\n",
    "    from kserve import V1beta1PredictorSpec\n",
    "    from kserve import V1beta1TFServingSpec\n",
    "    from datetime import datetime\n",
    "\n",
    "    namespace = utils.get_default_target_namespace()\n",
    "\n",
    "    now = datetime.now()\n",
    "    v = now.strftime(\"%Y-%m-%d--%H-%M-%S\")\n",
    "\n",
    "    name='digits-recognizer-{}'.format(v)\n",
    "    kserve_version='v1beta1'\n",
    "    api_version = constants.KSERVE_GROUP + '/' + kserve_version\n",
    "\n",
    "    isvc = V1beta1InferenceService(api_version=api_version,\n",
    "                                   kind=constants.KSERVE_KIND,\n",
    "                                   metadata=client.V1ObjectMeta(\n",
    "                                       name=name, namespace=namespace, annotations={'sidecar.istio.io/inject':'false'}),\n",
    "                                   spec=V1beta1InferenceServiceSpec(\n",
    "                                   predictor=V1beta1PredictorSpec(\n",
    "                                       service_account_name=\"sa-minio-kserve\",\n",
    "                                       tensorflow=(V1beta1TFServingSpec(\n",
    "                                           storage_uri=\"s3://mlpipeline/models/detect-digits/\"))))\n",
    "    )\n",
    "\n",
    "    KServe = KServeClient()\n",
    "    KServe.create(isvc)\n",
    "\n",
    "comp_get_data_batch = components.create_component_from_func(get_data_batch,base_image=\"public.ecr.aws/j1r0q0g6/notebooks/notebook-servers/jupyter-tensorflow-full:v1.5.0\")\n",
    "comp_get_latest_data = components.create_component_from_func(get_latest_data,base_image=\"public.ecr.aws/j1r0q0g6/notebooks/notebook-servers/jupyter-tensorflow-full:v1.5.0\")\n",
    "comp_reshape_data = components.create_component_from_func(reshape_data,base_image=\"public.ecr.aws/j1r0q0g6/notebooks/notebook-servers/jupyter-tensorflow-full:v1.5.0\")\n",
    "comp_model_building = components.create_component_from_func(model_building,base_image=\"public.ecr.aws/j1r0q0g6/notebooks/notebook-servers/jupyter-tensorflow-full:v1.5.0\")\n",
    "comp_model_serving = components.create_component_from_func(model_serving,base_image=\"public.ecr.aws/j1r0q0g6/notebooks/notebook-servers/jupyter-tensorflow-full:v1.5.0\",\n",
    "                                                           packages_to_install=['kserve==0.8.0.1'])\n",
    "\n",
    "\n",
    "@dsl.pipeline(\n",
    "    name='digits-recognizer-pipeline',\n",
    "    description='Detect digits'\n",
    ")\n",
    "def output_test(no_epochs,optimizer):\n",
    "    step1_1 = comp_get_data_batch()\n",
    "    step1_2 = comp_get_latest_data()\n",
    "    \n",
    "    step2 = comp_reshape_data()\n",
    "    step2.after(step1_1)\n",
    "    step2.after(step1_2)\n",
    "    \n",
    "    step3 = comp_model_building(no_epochs,optimizer)\n",
    "    step3.after(step2)\n",
    "    \n",
    "    step4 = comp_model_serving()\n",
    "    step4.after(step3)\n",
    "\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "    client = kfp.Client()\n",
    "\n",
    "    arguments = {\n",
    "        \"no_epochs\" : 1,\n",
    "        \"optimizer\": \"adam\"\n",
    "    }\n",
    "\n",
    "    run_directly = 1\n",
    "    \n",
    "    if (run_directly == 1):\n",
    "        client.create_run_from_pipeline_func(output_test,arguments=arguments,experiment_name=\"test\")\n",
    "    else:\n",
    "        kfp.compiler.Compiler().compile(pipeline_func=output_test,package_path='output_test.yaml')\n",
    "        client.upload_pipeline_version(pipeline_package_path='output_test.yaml',pipeline_version_name=\"0.4\",pipeline_name=\"pipeline test\",description=\"just for testing\")"
   ]
  }
 ],
 "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.8.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}