ML/notebooks/Prediction-of-Room-Occupancy-Synapse-Pub.ipynb

{
  "cells": [
    {
      "cell_type": "markdown",
      "source": [
        "# Train your Model on Spark, score it on Data Explorer pool\r\n",
        "\r\n",
        "<img src=\"https://github.com/Azure/azure-kusto-spark/raw/master/kusto_spark.png\" style=\"border: 1px solid #aaa; border-radius: 10px 10px 10px 10px; box-shadow: 5px 5px 5px #aaa\"/>\r\n",
        "\r\n",
        "In many use cases Machine Learning models are built and applied over data that is stored and managed by Azure Data Explorer (ADX). Most ML models are built and deployed in two steps:\r\n",
        "\r\n",
        "* Offline training\r\n",
        "* Real time scoring\r\n",
        "\r\n",
        "\r\n",
        "ML Training is a long and iterative process. Commonly, a model is developed by researchers/data scientists. They fetch the training data, clean it, engineer features, try different models and tune parameters, \r\n",
        "repeating this cycle until the ML model meets the required accuracy and robustness. To improve accuracy, they can:\r\n",
        "\r\n",
        "* Use a big data set, if available, that might contain hundreds of millions of records and plenty of features (dimensions and metrics)\r\n",
        "* Train very complex models, e.g., a DNN with dozens of layers and millions of free parameters\r\n",
        "* Perform a more exhaustive search for tuning the model’s hyper parameters\r\n",
        "Once the model is ready, it can be deployed to production for scoring.\r\n",
        "\r\n",
        "\r\n",
        "ML Scoring is the process of applying the model on new data to get predictions/regressions. Scoring usually needs to be done with minimal latency (near real time) for batches of streamed data.\r\n",
        "\r\n",
        " \r\n",
        "Data Explorer supports running inline Python scripts that are embedded in the KQL query. The Python code runs on the existing compute nodes of Data Explorer Pools, in a distributed manner near the data. \r\n",
        "It can handle Data Frames containing many millions of records, partitioned and processed on multiple nodes. This optimized architecture results in great performance and minimal latency.\r\n",
        "\r\n",
        "Specifically, for ML workloads, Data Explorer can be used for both training and scoring:\r\n",
        "\r\n",
        "* Scoring on ADX is the ultimate solution for data that is stored on ADX, as\r\n",
        "  * Processing is done near the data, which guarantees the fastest performance\r\n",
        "  * Embedding the scoring Python code in KQL query is simple, robust and cheap, relative to the usage of an external scoring service that requires management, networking, security, etc.\r\n",
        "Scoring can be done using the predict_fl() library function\r\n",
        "\r\n",
        "* Training on ADX can be done in case the full training data set is stored in Data Explorer, the training process takes up to few minutes and doesn’t require GPUs or other special hardware\r\n",
        "Still in many scenarios training is done on Big Data systems, such as Spark. Specifically, ML training on these systems is preferred in case that:\r\n",
        "\r\n",
        "* The training data is not stored in ADX, but in the data lake or other external storage/db\r\n",
        "* The training process is long (takes more than 5-10 minutes), usually done in batch/async mode\r\n",
        "\r\n",
        "Training can be accelerated by using GPUs\r\n",
        "Data Explorer production workflows must not be compromised by lengthy, CPU intensive, training jobs.\r\n",
        "So we end up in a workflow that uses Spark for training, and Data Explorer for scoring. \r\n",
        "Training on Spark is mostly done using the Spark ML framework, that is optimized for Spark architecture, but not supported by plain vanilla Python environment like ADX Python. So how can we still score in ADX?\r\n",
        "\r\n",
        "The solution is built from these steps:\r\n",
        "\r\n",
        "1. Fetch the training data from Data Explorer to Spark \r\n",
        "1. Train an ML model in Spark\r\n",
        "1. Convert the model to ONNX\r\n",
        "1. Serialize and export the model to Data Explorer \r\n",
        "1. Score in ADX using onnxruntime\r\n",
        "\r\n",
        "**Prerequisites:**\r\n",
        "\r\n",
        "Enable the Python plugin on your Data Explorer Pool and load the OccupancyDetection dataset to your Data Explorer Database (see the \r\n",
        "[KQL script](https://ms.web.azuresynapse.net/authoring/analyze/kqlscripts/Prediction-of-Room-Occupancy?feature.useKustoClientCanary=true&feature.useKustoManagement=true&useKql=true&showKustoPool=true&workspace=%2Fsubscriptions%2F2e131dbf-96b3-4377-9c8e-de5d3047f566%2FresourceGroups%2FDEMO%2Fproviders%2FMicrosoft.Synapse%2Fworkspaces%2Fcontosoworkspace01)).\r\n",
        "\r\n",
        "In the following example we build a logistic regression model to predict room occupancy based on Occupancy Detection data, a public dataset from UCI Repository. \r\n",
        "This model is a binary classifier to predict occupied or empty rooms based on temperature, humidity, light and CO2 sensors measurements. \r\n",
        "The example contains the full process of retrieving the data from ADX, building the model, convert it to ONNX and push it to your Data Explorer Pool. \r\n",
        "\r\n",
        "Finally the KQL scoring query can be run using a KQL script."
      ],
      "metadata": {
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "markdown",
      "source": [
        "## 1. Load the data from ADX to Synapse Spark"
      ],
      "metadata": {
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "code",
      "source": [
        "pyKusto = spark.builder.appName(\"kustoPySpark\").getOrCreate()"
      ],
      "outputs": [],
      "execution_count": null,
      "metadata": {
        "jupyter": {
          "source_hidden": false,
          "outputs_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "code",
      "source": [
        "# Read the data from the Data Explorer table \r\n",
        "crp = sc._jvm.com.microsoft.azure.kusto.data.ClientRequestProperties()\r\n",
        "crp.setOption(\"norequesttimeout\",True)\r\n",
        "crp.toString()\r\n",
        "\r\n",
        "df  = spark.read \\\r\n",
        "            .format(\"com.microsoft.kusto.spark.synapse.datasource\") \\\r\n",
        "            .option(\"spark.synapse.linkedService\", \"<DX pool linked service>\") \\\r\n",
        "            .option(\"kustoDatabase\", \"<DX pool DB>\") \\\r\n",
        "            .option(\"kustoQuery\", \"OccupancyDetection\") \\\r\n",
        "            .option(\"clientRequestPropertiesJson\", crp.toString()) \\\r\n",
        "            .option(\"readMode\", 'ForceDistributedMode') \\\r\n",
        "            .load()\r\n"
      ],
      "outputs": [],
      "execution_count": null,
      "metadata": {
        "jupyter": {
          "source_hidden": false,
          "outputs_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "code",
      "source": [
        "df.take(4)"
      ],
      "outputs": [],
      "execution_count": null,
      "metadata": {
        "jupyter": {
          "source_hidden": false,
          "outputs_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "code",
      "source": [
        "df.groupBy('Test', 'Occupancy').count().show()"
      ],
      "outputs": [],
      "execution_count": null,
      "metadata": {
        "jupyter": {
          "source_hidden": false,
          "outputs_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "markdown",
      "source": [
        "## 2. Train the ML model in Synapse Spark"
      ],
      "metadata": {
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "code",
      "source": [
        "s_train_x = df.filter(df.Test == False).select('Temperature', 'Humidity', 'Light', 'CO2', 'HumidityRatio')\r\n",
        "s_train_y = df.filter(df.Test == False).select('Occupancy')\r\n",
        "s_test_x = df.filter(df.Test == True).select('Temperature', 'Humidity', 'Light', 'CO2', 'HumidityRatio')\r\n",
        "s_test_y = df.filter(df.Test == True).select('Occupancy')"
      ],
      "outputs": [],
      "execution_count": null,
      "metadata": {
        "jupyter": {
          "source_hidden": false,
          "outputs_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "code",
      "source": [
        "df = df.withColumn('Label', df['Occupancy'].cast('int'))\r\n",
        "df.printSchema()"
      ],
      "outputs": [],
      "execution_count": null,
      "metadata": {
        "jupyter": {
          "source_hidden": false,
          "outputs_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "code",
      "source": [
        "df.show(4)"
      ],
      "outputs": [],
      "execution_count": null,
      "metadata": {
        "jupyter": {
          "source_hidden": false,
          "outputs_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "code",
      "source": [
        "s_train = df.filter(df.Test == False)\r\n",
        "s_test = df.filter(df.Test == True)\r\n",
        "print(s_train.count(), s_test.count())\r\n",
        "s_train.take(4)"
      ],
      "outputs": [],
      "execution_count": null,
      "metadata": {
        "jupyter": {
          "source_hidden": false,
          "outputs_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "code",
      "source": [
        "# Prepare the input for the model\r\n",
        "\r\n",
        "# Spark Logistic Regression estimator requires integer label so create it from the boolean Occupancy column\r\n",
        "df = df.withColumn('Label', df['Occupancy'].cast('int'))\r\n",
        "\r\n",
        "# Split to train & test sets\r\n",
        "s_train = df.filter(df.Test == False)\r\n",
        "s_test = df.filter(df.Test == True)\r\n",
        "\r\n",
        "# Create the Logistic Regression model\r\n",
        "from pyspark.ml.feature import VectorAssembler\r\n",
        "from pyspark.ml.classification import LogisticRegression\r\n",
        "\r\n",
        "# The Logistic Regression estimator expects the features in a single column so create it using VectorAssembler\r\n",
        "features = ('Temperature', 'Humidity', 'Light', 'CO2', 'HumidityRatio')\r\n",
        "assembler = VectorAssembler(inputCols=features,outputCol='Features')\r\n",
        "s_train_features = assembler.transform(s_train)\r\n",
        "s_train_features.take(4)\r\n",
        "lr = LogisticRegression(labelCol='Label', featuresCol='Features',maxIter=10)\r\n",
        "s_clf = lr.fit(s_train_features)\r\n",
        "\r\n",
        "# Predict the training set\r\n",
        "s_predict_train = s_clf.transform(s_train_features)\r\n",
        "\r\n",
        "# Predict the testing set\r\n",
        "s_test_features = assembler.transform(s_test)\r\n",
        "s_predict_test = s_clf.transform(s_test_features)\r\n",
        "s_predict_test.select(['Timestamp', 'Features', 'Label', 'prediction']).show(10)\r\n"
      ],
      "outputs": [],
      "execution_count": null,
      "metadata": {
        "jupyter": {
          "source_hidden": false,
          "outputs_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "code",
      "source": [
        "s_test_features = assembler.transform(s_test)\r\n",
        "s_test_features.take(4)\r\n",
        "s_predict_test = s_clf.transform(s_test_features)\r\n",
        "s_predict_test.show(10)"
      ],
      "outputs": [],
      "execution_count": null,
      "metadata": {
        "jupyter": {
          "source_hidden": false,
          "outputs_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "code",
      "source": [
        "# Calculate accuracy on the testing set\r\n",
        "\r\n",
        "import pyspark.sql.functions as F\r\n",
        "check = s_predict_test.withColumn('correct', F.when(F.col('Label') == F.col('prediction'), 1).otherwise(0))\r\n",
        "check.groupby('correct').count().show()\r\n",
        "accuracy = check.filter(check['correct'] == 1).count()/check.count()*100"
      ],
      "outputs": [],
      "execution_count": null,
      "metadata": {
        "jupyter": {
          "source_hidden": false,
          "outputs_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "markdown",
      "source": [
        " ## 3. Convert the model to ONNX"
      ],
      "metadata": {
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "code",
      "source": [
        "from onnxmltools import convert_sparkml\r\n",
        "from onnxmltools.convert.sparkml.utils import FloatTensorType\r\n",
        "\r\n",
        "initial_types = [('Features', FloatTensorType([None, 5]))]\r\n",
        "onnx_model = convert_sparkml(s_clf, 'Occupancy detection Pyspark Logistic Regression model', initial_types, spark_session = pyKusto)\r\n",
        "onnx_model"
      ],
      "outputs": [],
      "execution_count": null,
      "metadata": {
        "jupyter": {
          "source_hidden": false,
          "outputs_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "markdown",
      "source": [
        "## 4. Export the model to ADX"
      ],
      "metadata": {
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "code",
      "source": [
        "import datetime\r\n",
        "import pandas as pd\r\n",
        "\r\n",
        "smodel = onnx_model.SerializeToString().hex()\r\n",
        "models_tbl = 'ML_Models'\r\n",
        "model_name = 'Occupancy_Detection_LR'\r\n",
        "\r\n",
        "# Create a DataFrame containing a single row with model name, training time and\r\n",
        "# the serialized model, to be appended to the models table\r\n",
        "now = datetime.datetime.now()\r\n",
        "dfm = pd.DataFrame({'name':[model_name], 'timestamp':[now], 'model':[smodel]})\r\n",
        "sdfm = spark.createDataFrame(dfm)\r\n",
        "sdfm.show()"
      ],
      "outputs": [],
      "execution_count": null,
      "metadata": {
        "jupyter": {
          "source_hidden": false,
          "outputs_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "code",
      "source": [
        "extentsCreationTime = sc._jvm.org.joda.time.DateTime.now()\r\n",
        "\r\n",
        "sp = sc._jvm.com.microsoft.kusto.spark.datasink.SparkIngestionProperties(\r\n",
        "        True, None, None, None, None, extentsCreationTime, None, None)"
      ],
      "outputs": [],
      "execution_count": null,
      "metadata": {
        "jupyter": {
          "source_hidden": false,
          "outputs_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "code",
      "source": [
        "# Write the model to Data Explorer\r\n",
        "sdfm.write.format(\"com.microsoft.kusto.spark.synapse.datasource\") \\\r\n",
        ".option(\"spark.synapse.linkedService\", \"<DX pool linked service>\") \\\r\n",
        ".option(\"kustoDatabase\", \"<DX pool DB>\") \\\r\n",
        ".option(\"kustoTable\", models_tbl) \\\r\n",
        ".option(\"sparkIngestionPropertiesJson\", sp.toString()) \\\r\n",
        ".option(\"tableCreateOptions\",\"CreateIfNotExist\") \\\r\n",
        ".mode(\"Append\") \\\r\n",
        ".save()"
      ],
      "outputs": [],
      "execution_count": null,
      "metadata": {
        "jupyter": {
          "source_hidden": false,
          "outputs_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "markdown",
      "source": [
        "## 5. Score in ADX\r\n",
        "Is done by calling predict_onnx_fl() You can either install this function in your database, or call it in ad-hoc manner:\r\n"
      ],
      "metadata": {
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "code",
      "source": [
        "prediction_query = '''\r\n",
        "let predict_onnx_fl=(samples:(*), models_tbl:(name:string, timestamp:datetime, model:string), model_name:string, features_cols:dynamic, pred_col:string)\r\n",
        "{\r\n",
        "\r\n",
        "    let model_str = toscalar(models_tbl | where name == model_name | top 1 by timestamp desc | project model);\r\n",
        "    let kwargs = pack('smodel', model_str, 'features_cols', features_cols, 'pred_col', pred_col);\r\n",
        "    let code = ```if 1:\r\n",
        "    import binascii\r\n",
        "    smodel = kargs[\"smodel\"]\r\n",
        "    features_cols = kargs[\"features_cols\"]\r\n",
        "    pred_col = kargs[\"pred_col\"]\r\n",
        "    bmodel = binascii.unhexlify(smodel)\r\n",
        "    features_cols = kargs[\"features_cols\"]\r\n",
        "    pred_col = kargs[\"pred_col\"]\r\n",
        "    import onnxruntime as rt\r\n",
        "    sess = rt.InferenceSession(bmodel)\r\n",
        "    input_name = sess.get_inputs()[0].name\r\n",
        "    label_name = sess.get_outputs()[0].name\r\n",
        "    df1 = df[features_cols]\r\n",
        "    predictions = sess.run([label_name], {input_name: df1.values.astype(np.float32)})[0]\r\n",
        "    result = df\r\n",
        "    result[pred_col] = pd.DataFrame(predictions, columns=[pred_col])\r\n",
        "    ```;\r\n",
        "    samples | evaluate python(typeof(*), code, kwargs)\r\n",
        "};\r\n",
        "OccupancyDetection \r\n",
        "| where Test == 1\r\n",
        "| extend pred_Occupancy=int(null)\r\n",
        "| invoke predict_onnx_fl(ML_Models, 'Occupancy_Detection_LR', pack_array('Temperature', 'Humidity', 'Light', 'CO2', 'HumidityRatio'), 'pred_Occupancy')\r\n",
        "| summarize correct = countif(Occupancy == pred_Occupancy), incorrect = countif(Occupancy != pred_Occupancy), total = count()\r\n",
        "| extend accuracy = 100.0*correct/total\r\n",
        "'''"
      ],
      "outputs": [],
      "execution_count": null,
      "metadata": {
        "jupyter": {
          "source_hidden": false,
          "outputs_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "code",
      "source": [
        "spdf  = spark.read \\\r\n",
        "            .format(\"com.microsoft.kusto.spark.synapse.datasource\") \\\r\n",
        "            .option(\"spark.synapse.linkedService\", \"<DX pool linked service>\") \\\r\n",
        "            .option(\"kustoDatabase\", \"<DX pool DB>\") \\\r\n",
        "            .option(\"kustoQuery\", prediction_query) \\\r\n",
        "            .option(\"clientRequestPropertiesJson\", crp.toString()) \\\r\n",
        "            .option(\"readMode\", 'ForceDistributedMode') \\\r\n",
        "            .load()\r\n",
        "spdf.show ()"
      ],
      "outputs": [],
      "execution_count": null,
      "metadata": {
        "jupyter": {
          "source_hidden": false,
          "outputs_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    },
    {
      "cell_type": "markdown",
      "source": [],
      "metadata": {
        "nteract": {
          "transient": {
            "deleting": false
          }
        }
      }
    }
  ],
  "metadata": {
    "kernelspec": {
      "name": "synapse_pyspark",
      "display_name": "Synapse PySpark"
    },
    "language_info": {
      "name": "python"
    },
    "description": null,
    "save_output": true,
    "synapse_widget": {
      "version": "0.1",
      "state": {}
    }
  },
  "nbformat": 4,
  "nbformat_minor": 2
}