chapter_11_MLModelManagement.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "e6360fe1",
   "metadata": {},
   "source": [
    "# Chapter 11: Managing, Development, and Scaling Machine Learning Pipelines\n",
    "Christoph Windheuser    \n",
    "July, 2022   \n",
    "Python examples of chapter 11 (page 323 ff) in the book *Learning Spark*"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4cbf4435",
   "metadata": {},
   "source": [
    "## Installing MLflow\n",
    "To run these code, you first have to install MLflow. Run the followinf code in your terminal on your local machine:\n",
    "````\n",
    "pip install mlflow\n",
    "````\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "0d276868",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Import required python spark libraries\n",
    "import pyspark\n",
    "from pyspark.ml import Pipeline\n",
    "from pyspark.ml.feature import StringIndexer, VectorAssembler\n",
    "from pyspark.ml.regression import RandomForestRegressor\n",
    "from pyspark.ml.evaluation import RegressionEvaluator\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "55bbf54c",
   "metadata": {},
   "outputs": [],
   "source": [
    "# create a SparkSession\n",
    "# This requires access to the internet. If executed offline, an error is thrown\n",
    "\n",
    "spark = (SparkSession \\\n",
    "         .builder \\\n",
    "         .appName(\"Chapter_11\") \\\n",
    "         .getOrCreate())\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "923d0bf7",
   "metadata": {},
   "outputs": [],
   "source": [
    "filePath = \"../DB_Spark/LearningSparkV2/databricks-datasets/learning-spark-v2/sf-airbnb/sf-airbnb-clean.parquet\"\n",
    "airbnbDF = spark.read.parquet(filePath)\n",
    "\n",
    "(trainDF, testDF) = airbnbDF.randomSplit([0.8, 0.2], seed=42)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "620b6639",
   "metadata": {},
   "outputs": [],
   "source": [
    "categoricalCols = [field for (field, dataType) in trainDF.dtypes if dataType == \"string\"]\n",
    "indexOutputCols = [x + \"Index\" for x in categoricalCols]\n",
    "stringIndexer   = StringIndexer(inputCols = categoricalCols,\n",
    "                                outputCols = indexOutputCols,\n",
    "                                handleInvalid=\"skip\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "5cc6d91a",
   "metadata": {},
   "outputs": [],
   "source": [
    "numericCols     = [field for (field, dataType) in trainDF.dtypes\n",
    "                   if ((dataType == \"double\") & (field != \"price\"))]\n",
    "assemblerInputs = indexOutputCols + numericCols\n",
    "vecAssembler    = VectorAssembler(inputCols=assemblerInputs, outputCol=\"features\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "5438af8b",
   "metadata": {},
   "outputs": [],
   "source": [
    "rf = RandomForestRegressor(labelCol=\"price\", maxBins=40, maxDepth=7, numTrees=150, seed=42)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "43ff0a93",
   "metadata": {},
   "outputs": [],
   "source": [
    "pipeline = Pipeline(stages=[stringIndexer, vecAssembler, rf])"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "33065b87",
   "metadata": {},
   "source": [
    "# Run MLflow"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "bbecbc84",
   "metadata": {},
   "outputs": [],
   "source": [
    "import mlflow\n",
    "import mlflow.spark\n",
    "import pandas as pd"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "0dc68218",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Start MLflow:\n",
    "run = mlflow.start_run(run_name=\"random_forest\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "89898cd8",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Log params:\n",
    "mlflow.log_param(\"num_trees\", rf.getNumTrees())\n",
    "mlflow.log_param(\"max_depth\", rf.getMaxDepth())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "17bf578c",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Run model\n",
    "pipelineModel = pipeline.fit(trainDF)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "d7e76c3f",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2022/07/26 21:20:33 WARNING mlflow.utils.environment: Encountered an unexpected error while inferring pip requirements (model URI: /var/folders/1q/ycrkhm_50xd1x94vc0dv4c4m6zsl2_/T/tmp5x_f3fzb/model, flavor: spark), fall back to return ['pyspark==3.2.1']. Set logging level to DEBUG to see the full traceback.\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "ModelInfo(artifact_path='model', flavors={'spark': {'pyspark_version': '3.2.1', 'model_data': 'sparkml', 'code': None}, 'python_function': {'loader_module': 'mlflow.spark', 'python_version': '3.8.8', 'data': 'sparkml', 'env': 'conda.yaml'}}, model_uri='runs:/269124bda55c492c8ab2872196271497/model', model_uuid='82fac650d7234bd4a5e5f0a99d91e519', run_id='269124bda55c492c8ab2872196271497', saved_input_example_info=None, signature_dict=None, utc_time_created='2022-07-26 19:19:54.669919', mlflow_version='1.27.0')"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Log model:\n",
    "mlflow.spark.log_model(pipelineModel, \"model\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "895a0a8d",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Log metrics: RMSE and R2\n",
    "predDF = pipelineModel.transform(testDF)\n",
    "regressionEvaluator = RegressionEvaluator(predictionCol = \"prediction\",\n",
    "                                          labelCol      = \"price\")\n",
    "rmse = regressionEvaluator.setMetricName(\"rmse\").evaluate(predDF)\n",
    "r2   = regressionEvaluator.setMetricName(\"r2\").evaluate(predDF)\n",
    "mlflow.log_metrics({\"rmse\": rmse, \"r2\": r2})\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "ffdea53e",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Log artifacts: Feature importance scores\n",
    "rfModel  = pipelineModel.stages[-1]\n",
    "pandasDF = (pd.DataFrame(list(zip(vecAssembler.getInputCols(),\n",
    "                                  rfModel.featureImportances)),\n",
    "                         columns=[\"feature\", \"importance\"])\n",
    "            .sort_values(by=\"importance\", ascending=False))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "47d78f40",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Write to local file system, then tell MLflow where to find that file\n",
    "pandasDF.to_csv(\"feature-importance.csv\", index=False)\n",
    "mlflow.log_artifact(\"feature-importance.csv\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "31ff0c24",
   "metadata": {},
   "source": [
    "### Display MLflow in your browser\n",
    "run the following comand in your terminal:     \n",
    "````\n",
    "mlflow ui\n",
    "````\n",
    "Then navigate in your browser to http://localhost:5000/ (or http://127.0.0.1:5000/) to see the output of MLflow."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b895c8cc",
   "metadata": {},
   "source": [
    "### Query the MLflow tracking server by an API (MlflowClient)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "31b731a7",
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'r2': 0.2097921402566152, 'rmse': 213.9814486105752}"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from mlflow.tracking import MlflowClient\n",
    "\n",
    "client = MlflowClient()\n",
    "runs   = client.search_runs(run.info.experiment_id,\n",
    "                            order_by=[\"attributes.start_time desc\"],\n",
    "                            max_results=1)\n",
    "\n",
    "run_id = runs[0].info.run_id\n",
    "runs[0].data.metrics\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9cb36341",
   "metadata": {},
   "source": [
    "# Running experiments in the command line\n",
    "Experiments from a github repo can be run from the command line. The results are automatically incluced in MLflow.    \n",
    "Run the folloing command in your terminal:\n",
    "````\n",
    "mlflow run https://github.com/databricks/LearningSparkV2/#mlflow-project-example\n",
    "-P max_depth=5 -P num_trees=100\n",
    "````\n",
    "After finishing the run, you see the results of the run in your MLflow UI in the browser."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2b7886d0",
   "metadata": {},
   "source": [
    "# Stopping MLflow"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "d93aff89",
   "metadata": {},
   "outputs": [],
   "source": [
    "mlflow.end_run()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "df68ce12",
   "metadata": {},
   "source": [
    "# Model Deployment Options with MLlib\n",
    "Page 330 ff"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8e498b2b",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}