modeling.ipynb

{
 "cells": [
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Machine Learning with Spark on Python (PySpark) and MLflow for model tracking\n",
    "\n",
    "Here you will find pipeline implementation on PySpark and MLflow tracking for production. We will use a basic dataset called saber 11 that has been previously saved as delta and as table on local datalake, see `etl.py` file for more details."
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Libraries to use"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import warnings\n",
    "from pyspark import SparkContext\n",
    "from pyspark.sql import *\n",
    "from pyspark.sql.functions import *\n",
    "from pyspark.sql.types import *\n",
    "from pyspark.ml.feature import VectorAssembler\n",
    "from pyspark.ml.feature import OneHotEncoder, StringIndexer, MinMaxScaler\n",
    "from pyspark.ml.evaluation import BinaryClassificationEvaluator\n",
    "from pyspark.ml import Pipeline\n",
    "from pyspark.ml.classification import LinearSVC, LogisticRegression, DecisionTreeClassifier, RandomForestClassifier\n",
    "from pyspark.ml.tuning import ParamGridBuilder, CrossValidator\n",
    "from sklearn.metrics import recall_score, precision_score, accuracy_score\n",
    "import matplotlib.pyplot as plt\n",
    "\n",
    "warnings.filterwarnings('ignore')\n",
    "os.environ[\"JAVA_HOME\"] = \"/usr/lib/jvm/java-11-openjdk-amd64\"\n",
    "os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages  org.apache.spark:spark-avro_2.12:3.4.1,io.delta:delta-core_2.12:2.4.0 pyspark-shell'"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Spark session"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "your 131072x1 screen size is bogus. expect trouble\n",
      "23/07/11 22:31:51 WARN Utils: Your hostname, DESKTOP-JKDOQO9 resolves to a loopback address: 127.0.1.1; using 172.26.19.142 instead (on interface eth0)\n",
      "23/07/11 22:31:51 WARN Utils: Set SPARK_LOCAL_IP if you need to bind to another address\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      ":: loading settings :: url = jar:file:/home/josealcocer27/.local/lib/python3.8/site-packages/pyspark/jars/ivy-2.5.1.jar!/org/apache/ivy/core/settings/ivysettings.xml\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Ivy Default Cache set to: /home/josealcocer27/.ivy2/cache\n",
      "The jars for the packages stored in: /home/josealcocer27/.ivy2/jars\n",
      "org.apache.spark#spark-avro_2.12 added as a dependency\n",
      "io.delta#delta-core_2.12 added as a dependency\n",
      ":: resolving dependencies :: org.apache.spark#spark-submit-parent-f3036e09-3374-42c8-aac8-5279bbb8f09d;1.0\n",
      "\tconfs: [default]\n",
      "\tfound org.apache.spark#spark-avro_2.12;3.4.1 in central\n",
      "\tfound org.tukaani#xz;1.9 in central\n",
      "\tfound io.delta#delta-core_2.12;2.4.0 in central\n",
      "\tfound io.delta#delta-storage;2.4.0 in central\n",
      "\tfound org.antlr#antlr4-runtime;4.9.3 in central\n",
      ":: resolution report :: resolve 266ms :: artifacts dl 14ms\n",
      "\t:: modules in use:\n",
      "\tio.delta#delta-core_2.12;2.4.0 from central in [default]\n",
      "\tio.delta#delta-storage;2.4.0 from central in [default]\n",
      "\torg.antlr#antlr4-runtime;4.9.3 from central in [default]\n",
      "\torg.apache.spark#spark-avro_2.12;3.4.1 from central in [default]\n",
      "\torg.tukaani#xz;1.9 from central in [default]\n",
      "\t---------------------------------------------------------------------\n",
      "\t|                  |            modules            ||   artifacts   |\n",
      "\t|       conf       | number| search|dwnlded|evicted|| number|dwnlded|\n",
      "\t---------------------------------------------------------------------\n",
      "\t|      default     |   5   |   0   |   0   |   0   ||   5   |   0   |\n",
      "\t---------------------------------------------------------------------\n",
      ":: retrieving :: org.apache.spark#spark-submit-parent-f3036e09-3374-42c8-aac8-5279bbb8f09d\n",
      "\tconfs: [default]\n",
      "\t0 artifacts copied, 5 already retrieved (0kB/9ms)\n",
      "23/07/11 22:31:53 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n",
      "Setting default log level to \"WARN\".\n",
      "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n"
     ]
    }
   ],
   "source": [
    "spark = SparkSession.builder.master(\"local[*]\").appName(\"Spark_ML\") \\\n",
    "        .config(\"spark.sql.extensions\", \"io.delta.sql.DeltaSparkSessionExtension\") \\\n",
    "        .config(\"spark.sql.catalog.spark_catalog\", \"org.apache.spark.sql.delta.catalog.DeltaCatalog\")\\\n",
    "        .enableHiveSupport()\\\n",
    "        .getOrCreate()\n",
    "\n",
    "sc = SparkContext.getOrCreate()\n",
    "sqlCtx = SQLContext(sc)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Data reading and basic EDA"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = spark.read.format('delta')\\\n",
    "    .load('delta/saber11')"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Number of rows and schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "23/07/11 22:32:15 WARN package: Truncated the string representation of a plan since it was too large. This behavior can be adjusted by setting 'spark.sql.debug.maxToStringFields'.\n",
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "10361"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- lenguaje: string (nullable = true)\n",
      " |-- matematicas: string (nullable = true)\n",
      " |-- sociales: string (nullable = true)\n",
      " |-- Id: integer (nullable = true)\n",
      " |-- filosofia: string (nullable = true)\n",
      " |-- biologia: string (nullable = true)\n",
      " |-- quimica: string (nullable = true)\n",
      " |-- fisica: string (nullable = true)\n",
      " |-- nivel_ingles: string (nullable = true)\n",
      " |-- sisben: integer (nullable = true)\n",
      " |-- estrato: integer (nullable = true)\n",
      " |-- genero: string (nullable = true)\n",
      " |-- puntaje_saber11: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Unique values at column lenguaje\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+\n",
      "|lenguaje|\n",
      "+--------+\n",
      "|    bajo|\n",
      "|   medio|\n",
      "|superior|\n",
      "|    alto|\n",
      "+--------+\n",
      "\n",
      "\n",
      "Unique values at column matematicas\n",
      "\n",
      "+-----------+\n",
      "|matematicas|\n",
      "+-----------+\n",
      "|       bajo|\n",
      "|      medio|\n",
      "|   superior|\n",
      "|       alto|\n",
      "+-----------+\n",
      "\n",
      "\n",
      "Unique values at column sociales\n",
      "\n",
      "+--------+\n",
      "|sociales|\n",
      "+--------+\n",
      "|    bajo|\n",
      "|   medio|\n",
      "|superior|\n",
      "|    alto|\n",
      "+--------+\n",
      "\n",
      "\n",
      "Unique values at column filosofia\n",
      "\n",
      "+---------+\n",
      "|filosofia|\n",
      "+---------+\n",
      "|     bajo|\n",
      "|    medio|\n",
      "| superior|\n",
      "|     alto|\n",
      "+---------+\n",
      "\n",
      "\n",
      "Unique values at column biologia\n",
      "\n",
      "+--------+\n",
      "|biologia|\n",
      "+--------+\n",
      "|    bajo|\n",
      "|   medio|\n",
      "|superior|\n",
      "|    alto|\n",
      "+--------+\n",
      "\n",
      "\n",
      "Unique values at column quimica\n",
      "\n",
      "+--------+\n",
      "| quimica|\n",
      "+--------+\n",
      "|    bajo|\n",
      "|   medio|\n",
      "|superior|\n",
      "|    alto|\n",
      "+--------+\n",
      "\n",
      "\n",
      "Unique values at column fisica\n",
      "\n",
      "+--------+\n",
      "|  fisica|\n",
      "+--------+\n",
      "|    bajo|\n",
      "|superior|\n",
      "|   medio|\n",
      "|    alto|\n",
      "+--------+\n",
      "\n",
      "\n",
      "Unique values at column nivel_ingles\n",
      "\n",
      "+------------+\n",
      "|nivel_ingles|\n",
      "+------------+\n",
      "|          A2|\n",
      "|          A-|\n",
      "|          B1|\n",
      "|          B+|\n",
      "|          A1|\n",
      "+------------+\n",
      "\n",
      "\n",
      "Unique values at column sisben\n",
      "\n",
      "+------+\n",
      "|sisben|\n",
      "+------+\n",
      "|     1|\n",
      "|     3|\n",
      "|    -9|\n",
      "|     5|\n",
      "|     4|\n",
      "|     2|\n",
      "+------+\n",
      "\n",
      "\n",
      "Unique values at column estrato\n",
      "\n",
      "+-------+\n",
      "|estrato|\n",
      "+-------+\n",
      "|      1|\n",
      "|      6|\n",
      "|      3|\n",
      "|      5|\n",
      "|      4|\n",
      "|      2|\n",
      "+-------+\n",
      "\n",
      "\n",
      "Unique values at column genero\n",
      "\n",
      "+------+\n",
      "|genero|\n",
      "+------+\n",
      "|Hombre|\n",
      "| Mujer|\n",
      "+------+\n",
      "\n",
      "\n",
      "Unique values at column puntaje_saber11\n",
      "\n",
      "+---------------+\n",
      "|puntaje_saber11|\n",
      "+---------------+\n",
      "|           bajo|\n",
      "|          medio|\n",
      "+---------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "for column in df.columns:\n",
    "    if column != 'Id':\n",
    "        print(f'\\nUnique values at column {column}\\n')\n",
    "        df.select(column).distinct().show()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Feature transform"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.drop('id')\n",
    "df = df.dropna()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Target to binary values"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.withColumn('puntaje_saber11',\n",
    "                   when(col('puntaje_saber11')=='bajo', 0)\\\n",
    "                   .otherwise(1))"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Balanced or imbalanced dataset?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------+-----+\n",
      "|puntaje_saber11|count|\n",
      "+---------------+-----+\n",
      "|              1| 7140|\n",
      "|              0| 3221|\n",
      "+---------------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupBy(\"puntaje_saber11\").count().show()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's proceed to balance the dataset"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------+-----+\n",
      "|puntaje_saber11|count|\n",
      "+---------------+-----+\n",
      "|              1| 3219|\n",
      "|              0| 3221|\n",
      "+---------------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "class_0 = df.groupBy(\"puntaje_saber11\").count().sort(asc(\"puntaje_saber11\")).collect()[0][1]\n",
    "class_1 = df.groupBy(\"puntaje_saber11\").count().sort(asc(\"puntaje_saber11\")).collect()[1][1]\n",
    "\n",
    "# Find the ratio of the class sizes\n",
    "class_ratio = class_0 / class_1 \n",
    "\n",
    "# Define the sampling fractions for each class\n",
    "sampling_fractions = {\n",
    "    0: 1.0,  # No sampling for class 0 (majority class)\n",
    "    1: class_ratio  # Sampling based on class ratio for class 1 (minority class)\n",
    "}\n",
    "\n",
    "balanced_df = df.sampleBy(\"puntaje_saber11\", fractions=sampling_fractions, seed=1234)\n",
    "balanced_df.groupBy(\"puntaje_saber11\").count().show()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### String Indexer, OneHotEncoder and MinMaxScaler"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "# features for OHE and MinMaxScaler\n",
    "\n",
    "OHE_cols = [x[0] for x in df.dtypes if x[1] == 'string' and x[0] != 'puntaje_saber11']\n",
    "to_scale = [x[0] for x in df.dtypes if x[1] in ['int', 'float'] and x[0] != 'puntaje_saber11']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------+-----------------------+--------------------------+-----------------------+------------------------+-----------------------+----------------------+---------------------+---------------------------+---------------------+--------------------+--------------------+\n",
      "|puntaje_saber11|lenguaje_encoded_scaled|matematicas_encoded_scaled|sociales_encoded_scaled|filosofia_encoded_scaled|biologia_encoded_scaled|quimica_encoded_scaled|fisica_encoded_scaled|nivel_ingles_encoded_scaled|genero_encoded_scaled|       sisben_scaled|      estrato_scaled|\n",
      "+---------------+-----------------------+--------------------------+-----------------------+------------------------+-----------------------+----------------------+---------------------+---------------------------+---------------------+--------------------+--------------------+\n",
      "|              0|          [1.0,0.0,0.0]|             [1.0,0.0,0.0]|          [1.0,0.0,0.0]|           [1.0,0.0,0.0]|          [0.0,1.0,0.0]|         [0.0,1.0,0.0]|            (3,[],[])|              (4,[0],[1.0])|                [1.0]|[0.7857142857142857]|               [0.4]|\n",
      "|              0|          [0.0,1.0,0.0]|                 (3,[],[])|          [0.0,1.0,0.0]|           [1.0,0.0,0.0]|          [0.0,1.0,0.0]|         [1.0,0.0,0.0]|        [0.0,1.0,0.0]|              (4,[0],[1.0])|                [1.0]|[0.7142857142857142]|               [0.0]|\n",
      "|              0|          [0.0,1.0,0.0]|             [0.0,1.0,0.0]|          [0.0,1.0,0.0]|           [0.0,1.0,0.0]|          [0.0,1.0,0.0]|         [0.0,1.0,0.0]|        [1.0,0.0,0.0]|              (4,[0],[1.0])|                [0.0]|               [1.0]|               [0.4]|\n",
      "|              0|          [0.0,1.0,0.0]|             [0.0,1.0,0.0]|          [0.0,1.0,0.0]|           [1.0,0.0,0.0]|          [0.0,1.0,0.0]|         [1.0,0.0,0.0]|        [0.0,1.0,0.0]|              (4,[2],[1.0])|                [1.0]|               [1.0]|[0.6000000000000001]|\n",
      "|              0|          [0.0,1.0,0.0]|             [0.0,1.0,0.0]|          [0.0,1.0,0.0]|           [1.0,0.0,0.0]|          [0.0,1.0,0.0]|         [0.0,1.0,0.0]|        [0.0,1.0,0.0]|              (4,[0],[1.0])|                [1.0]|[0.7142857142857142]|               [0.0]|\n",
      "+---------------+-----------------------+--------------------------+-----------------------+------------------------+-----------------------+----------------------+---------------------+---------------------------+---------------------+--------------------+--------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Multiple Indexers for every column in OHE_cols\n",
    "indexer = [StringIndexer(inputCol = col, outputCol = col+'_indexed') for col in OHE_cols]\n",
    "\n",
    "# Multiple OHE for every column in OHE_cols\n",
    "encoder = [OneHotEncoder(inputCol = col+'_indexed', outputCol = col+'_encoded') for col in OHE_cols]\n",
    "\n",
    "aux_features = [col+'_encoded' for col in OHE_cols] + to_scale\n",
    "\n",
    "# Vector Assembler to can apply MinMaxScaler\n",
    "vector_to_scale = [VectorAssembler(inputCols = [col], outputCol = col+'_vec') for col in aux_features]\n",
    "scaler = [MinMaxScaler(inputCol = col+'_vec', outputCol = col+'_scaled') for col in aux_features]\n",
    "\n",
    "# Final Vector Assembler\n",
    "vector = VectorAssembler(inputCols = [col+'_scaled' for col in aux_features], outputCol ='features')\n",
    "\n",
    "transform_steps = indexer + encoder + vector_to_scale + scaler\n",
    "\n",
    "# Pipeline\n",
    "features_pipeline = Pipeline(stages = transform_steps)\n",
    "df_trans = features_pipeline.fit(balanced_df).transform(balanced_df)\n",
    "to_select = [col for col in df_trans.columns if '_scaled' in col or col == 'puntaje_saber11']\n",
    "df_trans.select(to_select).show(5)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Modeling and MLflow tracking"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Train test split"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "(trainingData, testingData) = balanced_df.randomSplit([0.70, 0.3], seed = 1234)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Set uri and experiment in MLflow"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2023/07/11 22:32:45 INFO mlflow.store.db.utils: Creating initial MLflow database tables...\n",
      "2023/07/11 22:32:46 INFO mlflow.store.db.utils: Updating database tables\n",
      "INFO  [alembic.runtime.migration] Context impl SQLiteImpl.\n",
      "INFO  [alembic.runtime.migration] Will assume non-transactional DDL.\n",
      "INFO  [alembic.runtime.migration] Running upgrade  -> 451aebb31d03, add metric step\n",
      "INFO  [alembic.runtime.migration] Running upgrade 451aebb31d03 -> 90e64c465722, migrate user column to tags\n",
      "INFO  [alembic.runtime.migration] Running upgrade 90e64c465722 -> 181f10493468, allow nulls for metric values\n",
      "INFO  [alembic.runtime.migration] Running upgrade 181f10493468 -> df50e92ffc5e, Add Experiment Tags Table\n",
      "INFO  [alembic.runtime.migration] Running upgrade df50e92ffc5e -> 7ac759974ad8, Update run tags with larger limit\n",
      "INFO  [alembic.runtime.migration] Running upgrade 7ac759974ad8 -> 89d4b8295536, create latest metrics table\n",
      "INFO  [89d4b8295536_create_latest_metrics_table_py] Migration complete!\n",
      "INFO  [alembic.runtime.migration] Running upgrade 89d4b8295536 -> 2b4d017a5e9b, add model registry tables to db\n",
      "INFO  [2b4d017a5e9b_add_model_registry_tables_to_db_py] Adding registered_models and model_versions tables to database.\n",
      "INFO  [2b4d017a5e9b_add_model_registry_tables_to_db_py] Migration complete!\n",
      "INFO  [alembic.runtime.migration] Running upgrade 2b4d017a5e9b -> cfd24bdc0731, Update run status constraint with killed\n",
      "INFO  [alembic.runtime.migration] Running upgrade cfd24bdc0731 -> 0a8213491aaa, drop_duplicate_killed_constraint\n",
      "INFO  [alembic.runtime.migration] Running upgrade 0a8213491aaa -> 728d730b5ebd, add registered model tags table\n",
      "INFO  [alembic.runtime.migration] Running upgrade 728d730b5ebd -> 27a6a02d2cf1, add model version tags table\n",
      "INFO  [alembic.runtime.migration] Running upgrade 27a6a02d2cf1 -> 84291f40a231, add run_link to model_version\n",
      "INFO  [alembic.runtime.migration] Running upgrade 84291f40a231 -> a8c4a736bde6, allow nulls for run_id\n",
      "INFO  [alembic.runtime.migration] Running upgrade a8c4a736bde6 -> 39d1c3be5f05, add_is_nan_constraint_for_metrics_tables_if_necessary\n",
      "INFO  [alembic.runtime.migration] Running upgrade 39d1c3be5f05 -> c48cb773bb87, reset_default_value_for_is_nan_in_metrics_table_for_mysql\n",
      "INFO  [alembic.runtime.migration] Running upgrade c48cb773bb87 -> bd07f7e963c5, create index on run_uuid\n",
      "INFO  [alembic.runtime.migration] Running upgrade bd07f7e963c5 -> 0c779009ac13, add deleted_time field to runs table\n",
      "INFO  [alembic.runtime.migration] Running upgrade 0c779009ac13 -> cc1f77228345, change param value length to 500\n",
      "INFO  [alembic.runtime.migration] Running upgrade cc1f77228345 -> 97727af70f4d, Add creation_time and last_update_time to experiments table\n",
      "INFO  [alembic.runtime.migration] Running upgrade 97727af70f4d -> 3500859a5d39, Add Model Aliases table\n",
      "INFO  [alembic.runtime.migration] Running upgrade 3500859a5d39 -> 7f2a7d5fae7d, add datasets inputs input_tags tables\n",
      "INFO  [alembic.runtime.migration] Context impl SQLiteImpl.\n",
      "INFO  [alembic.runtime.migration] Will assume non-transactional DDL.\n",
      "2023/07/11 22:32:47 INFO mlflow.tracking.fluent: Experiment with name 'saber11' does not exist. Creating a new experiment.\n"
     ]
    }
   ],
   "source": [
    "import mlflow\n",
    "import mlflow.spark\n",
    "\n",
    "mlflow.set_tracking_uri(\"sqlite:///mlflow.db\")\n",
    "mlflow.set_experiment(experiment_name='saber11')\n",
    "mlflow.spark.autolog(disable=True)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "First let's create a function to create ROC curve that can be reused multiple times"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "def plot_roc(roc):\n",
    "    fig = plt.figure()\n",
    "    plt.plot(roc['FPR'], roc['TPR'])\n",
    "    plt.ylabel('False Positive Rate')\n",
    "    plt.xlabel('True Positive Rate')\n",
    "    plt.title('ROC Curve')\n",
    "    plt.close(fig)\n",
    "    return fig"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Baseline (logistic regression)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "23/07/11 22:33:00 WARN InstanceBuilder: Failed to load implementation from:dev.ludovic.netlib.blas.JNIBLAS\n",
      "2023/07/11 22:33:22 WARNING mlflow.utils.environment: Encountered an unexpected error while inferring pip requirements (model URI: /tmp/tmpkzcmwzmg/model, flavor: spark), fall back to return ['pyspark==3.4.1']. Set logging level to DEBUG to see the full traceback.\n"
     ]
    }
   ],
   "source": [
    "with mlflow.start_run(run_name = 'log_reg_baseline'):\n",
    "    # define model and set a tag\n",
    "    model_lr = LogisticRegression(featuresCol = 'features', labelCol = 'puntaje_saber11')\n",
    "    mlflow.set_tag('model_name', str(model_lr.__class__.__name__))\n",
    "\n",
    "    # define pipeline and fit\n",
    "    pipeline_lr = Pipeline(stages = [features_pipeline, vector, model_lr])\n",
    "    lr_model = pipeline_lr.fit(trainingData)\n",
    "    trainingSummary = lr_model.stages[-1].summary\n",
    "\n",
    "    # evaluate and log metrics and model\n",
    "    evaluator = BinaryClassificationEvaluator(rawPredictionCol = 'probability', labelCol = 'puntaje_saber11')\n",
    "    trainPredictions = lr_model.transform(trainingData)\n",
    "    trainAUC = evaluator.evaluate(trainPredictions)\n",
    "    testPredictions = lr_model.transform(testingData)\n",
    "    testAUC = evaluator.evaluate(testPredictions)\n",
    "    mlflow.log_metrics({\"train AUC\": trainAUC, \"test AUC\": testAUC, \"training accuracy\": trainingSummary.accuracy,\n",
    "                        \"training weighted precision\": trainingSummary.weightedPrecision, \n",
    "                        \"training weighted recall\": trainingSummary.weightedRecall})\n",
    "    mlflow.spark.log_model(lr_model, 'log_reg_spark')\n",
    "    roc = trainingSummary.roc.toPandas()\n",
    "    fig = plot_roc(roc)\n",
    "    mlflow.log_figure(fig, \"ROC_training_curve.png\")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Linear SVC"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "23/07/11 22:34:23 ERROR OWLQN: Failure! Resetting history: breeze.optimize.NaNHistory: \n",
      "23/07/11 22:34:34 ERROR OWLQN: Failure! Resetting history: breeze.optimize.NaNHistory: \n",
      "2023/07/11 22:37:09 WARNING mlflow.utils.environment: Encountered an unexpected error while inferring pip requirements (model URI: /tmp/tmpq76zbux1/model, flavor: spark), fall back to return ['pyspark==3.4.1']. Set logging level to DEBUG to see the full traceback.\n"
     ]
    }
   ],
   "source": [
    "with mlflow.start_run(run_name = 'Linear_SVC_with_grid'):    \n",
    "    # define model and set a tag\n",
    "    svc_model = LinearSVC(featuresCol='features', labelCol='puntaje_saber11')\n",
    "    mlflow.set_tag('model_name', str(svc_model.__class__.__name__))\n",
    "\n",
    "    # build parameter grid\n",
    "    paramGridSVM = ParamGridBuilder()\\\n",
    "        .addGrid(svc_model.regParam, [0.01, 0.5, 1, 5])\\\n",
    "        .addGrid(svc_model.maxIter, [100, 150])\\\n",
    "        .build()\n",
    "\n",
    "    # create an evaluator\n",
    "    evaluator = BinaryClassificationEvaluator(rawPredictionCol='rawPrediction', labelCol='puntaje_saber11')\n",
    "\n",
    "    # create cross validator for tuning\n",
    "    crossvalSVM = CrossValidator(estimator = svc_model,\n",
    "                                estimatorParamMaps = paramGridSVM,\n",
    "                                evaluator = evaluator,\n",
    "                                numFolds = 3)\n",
    "\n",
    "    # define pipeline with stages and train it\n",
    "    pipeline = Pipeline(stages=[features_pipeline, vector, crossvalSVM])\n",
    "    model = pipeline.fit(trainingData)\n",
    "\n",
    "    # evaluate and log metrics and model\n",
    "    trainPredictions = model.transform(trainingData)\n",
    "    trainAUC = evaluator.evaluate(trainPredictions)\n",
    "    testPredictions = model.transform(testingData)\n",
    "    testAUC = evaluator.evaluate(testPredictions)\n",
    "    best_model = model.stages[-1].bestModel\n",
    "    trainingSummary = best_model.summary()\n",
    "    mlflow.log_params(params={\"RegParam\": best_model.getRegParam(), \"maxIter\": best_model.getMaxIter()})\n",
    "    mlflow.log_metrics({\"train AUC\": trainAUC, \"test AUC\": testAUC, \"training accuracy\": trainingSummary.accuracy,\n",
    "                        \"training weighted precision\": trainingSummary.weightedPrecision, \n",
    "                        \"training weighted recall\": trainingSummary.weightedRecall})\n",
    "    new_pipeline =  Pipeline(stages=[features_pipeline, vector, best_model])\n",
    "    new_model = new_pipeline.fit(trainingData)\n",
    "    mlflow.spark.log_model(new_model, 'LinearSVC_Spark')\n",
    "    roc = trainingSummary.roc.toPandas()\n",
    "    fig = plot_roc(roc)\n",
    "    mlflow.log_figure(fig, f\"ROC_training_curve_{best_model.__class__.__name__}\"+\".png\")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Decision Tree\n",
    "\n",
    "- First define a function to math metrics because of some Spark's models don't have summary method"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "def create_metrics(spark_df):\n",
    "    df = spark_df.select('prediction', 'puntaje_saber11').toPandas()\n",
    "    for column in df.columns:\n",
    "        df[column] = df[column].astype('float64')\n",
    "    precision = precision_score(df.puntaje_saber11.values, df.prediction.values)\n",
    "    recall = recall_score(df.puntaje_saber11.values, df.prediction.values)\n",
    "    accuracy = accuracy_score(df.puntaje_saber11.values, df.prediction.values)\n",
    "    return precision, recall, accuracy"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2023/07/11 22:38:02 WARNING mlflow.utils.environment: Encountered an unexpected error while inferring pip requirements (model URI: /tmp/tmp_0fld5cu/model, flavor: spark), fall back to return ['pyspark==3.4.1']. Set logging level to DEBUG to see the full traceback.\n"
     ]
    }
   ],
   "source": [
    "with mlflow.start_run(run_name='decision_tree_with_grid'):\n",
    "    # define model and set a tag\n",
    "    tree_model = DecisionTreeClassifier(featuresCol='features', labelCol='puntaje_saber11')\n",
    "    mlflow.set_tag(\"model_name\", str(tree_model.__class__.__name__))\n",
    "\n",
    "    # build parameter grid\n",
    "    paramGridTree = ParamGridBuilder()\\\n",
    "        .addGrid(tree_model.maxDepth, [3, 5, 10])\\\n",
    "        .addGrid(tree_model.maxBins, [16, 32])\\\n",
    "        .build()\n",
    "    \n",
    "    # create an evaluator\n",
    "    evaluator = BinaryClassificationEvaluator(rawPredictionCol='probability', labelCol='puntaje_saber11')\n",
    "\n",
    "    # create cross validator\n",
    "    crossvalTree = CrossValidator(estimator=tree_model,\n",
    "                                  estimatorParamMaps=paramGridTree,\n",
    "                                  evaluator=evaluator,\n",
    "                                  numFolds=3)\n",
    "    \n",
    "    # define pipeline with stages and train it\n",
    "    pipeline = Pipeline(stages=[features_pipeline, vector, crossvalTree])\n",
    "    model = pipeline.fit(trainingData)\n",
    "\n",
    "    # evaluate and log metrics and model\n",
    "    trainPredictions = model.transform(trainingData)\n",
    "    trainAUC = evaluator.evaluate(trainPredictions)\n",
    "    testPredictions = model.transform(testingData)\n",
    "    testAUC = evaluator.evaluate(testPredictions)\n",
    "    best_model = model.stages[-1].bestModel\n",
    "    precision, recall, accuracy =  create_metrics(trainPredictions)\n",
    "    mlflow.log_params(params={\"maxDepth\": best_model.getMaxDepth(), \"maxBins\": best_model.getMaxBins()})\n",
    "    mlflow.log_metrics({\"train AUC\": trainAUC, \"test AUC\": testAUC, \"training accuracy\": accuracy,\n",
    "                    \"training weighted precision\": precision, \n",
    "                    \"training weighted recall\": recall})\n",
    "    new_pipeline =  Pipeline(stages=[features_pipeline, vector, best_model])\n",
    "    new_model = new_pipeline.fit(trainingData)\n",
    "    mlflow.spark.log_model(new_model, 'DecisionTree_Spark')"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Random Forest"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "23/07/11 22:38:21 WARN DAGScheduler: Broadcasting large task binary with size 1111.8 KiB\n",
      "23/07/11 22:38:22 WARN DAGScheduler: Broadcasting large task binary with size 1175.0 KiB\n",
      "23/07/11 22:38:25 WARN DAGScheduler: Broadcasting large task binary with size 1111.8 KiB\n",
      "23/07/11 22:38:25 WARN DAGScheduler: Broadcasting large task binary with size 1175.0 KiB\n",
      "23/07/11 22:38:26 WARN DAGScheduler: Broadcasting large task binary with size 1173.7 KiB\n",
      "23/07/11 22:38:27 WARN DAGScheduler: Broadcasting large task binary with size 1434.9 KiB\n",
      "23/07/11 22:38:27 WARN DAGScheduler: Broadcasting large task binary with size 1757.9 KiB\n",
      "23/07/11 22:38:27 WARN DAGScheduler: Broadcasting large task binary with size 1344.3 KiB\n",
      "23/07/11 22:38:28 WARN DAGScheduler: Broadcasting large task binary with size 1072.4 KiB\n",
      "23/07/11 22:38:28 WARN DAGScheduler: Broadcasting large task binary with size 1342.4 KiB\n",
      "23/07/11 22:38:29 WARN DAGScheduler: Broadcasting large task binary with size 1772.8 KiB\n",
      "23/07/11 22:38:29 WARN DAGScheduler: Broadcasting large task binary with size 2.3 MiB\n",
      "23/07/11 22:38:29 WARN DAGScheduler: Broadcasting large task binary with size 3.0 MiB\n",
      "23/07/11 22:38:30 WARN DAGScheduler: Broadcasting large task binary with size 2.1 MiB\n",
      "23/07/11 22:38:31 WARN DAGScheduler: Broadcasting large task binary with size 1111.8 KiB\n",
      "23/07/11 22:38:31 WARN DAGScheduler: Broadcasting large task binary with size 1457.1 KiB\n",
      "23/07/11 22:38:31 WARN DAGScheduler: Broadcasting large task binary with size 2025.2 KiB\n",
      "23/07/11 22:38:32 WARN DAGScheduler: Broadcasting large task binary with size 2.8 MiB\n",
      "23/07/11 22:38:32 WARN DAGScheduler: Broadcasting large task binary with size 4.0 MiB\n",
      "23/07/11 22:38:33 WARN DAGScheduler: Broadcasting large task binary with size 5.5 MiB\n",
      "23/07/11 22:38:34 WARN DAGScheduler: Broadcasting large task binary with size 3.6 MiB\n",
      "23/07/11 22:38:35 WARN DAGScheduler: Broadcasting large task binary with size 1173.7 KiB\n",
      "23/07/11 22:38:35 WARN DAGScheduler: Broadcasting large task binary with size 1434.9 KiB\n",
      "23/07/11 22:38:36 WARN DAGScheduler: Broadcasting large task binary with size 1757.9 KiB\n",
      "23/07/11 22:38:36 WARN DAGScheduler: Broadcasting large task binary with size 1344.3 KiB\n",
      "23/07/11 22:38:37 WARN DAGScheduler: Broadcasting large task binary with size 1072.4 KiB\n",
      "23/07/11 22:38:37 WARN DAGScheduler: Broadcasting large task binary with size 1342.3 KiB\n",
      "23/07/11 22:38:37 WARN DAGScheduler: Broadcasting large task binary with size 1772.8 KiB\n",
      "23/07/11 22:38:38 WARN DAGScheduler: Broadcasting large task binary with size 2.3 MiB\n",
      "23/07/11 22:38:38 WARN DAGScheduler: Broadcasting large task binary with size 3.0 MiB\n",
      "23/07/11 22:38:39 WARN DAGScheduler: Broadcasting large task binary with size 2.1 MiB\n",
      "23/07/11 22:38:40 WARN DAGScheduler: Broadcasting large task binary with size 1111.8 KiB\n",
      "23/07/11 22:38:40 WARN DAGScheduler: Broadcasting large task binary with size 1457.1 KiB\n",
      "23/07/11 22:38:40 WARN DAGScheduler: Broadcasting large task binary with size 2025.2 KiB\n",
      "23/07/11 22:38:41 WARN DAGScheduler: Broadcasting large task binary with size 2.8 MiB\n",
      "23/07/11 22:38:41 WARN DAGScheduler: Broadcasting large task binary with size 4.0 MiB\n",
      "23/07/11 22:38:42 WARN DAGScheduler: Broadcasting large task binary with size 5.5 MiB\n",
      "23/07/11 22:38:43 WARN DAGScheduler: Broadcasting large task binary with size 3.6 MiB\n",
      "23/07/11 22:38:54 WARN DAGScheduler: Broadcasting large task binary with size 1092.3 KiB\n",
      "23/07/11 22:38:55 WARN DAGScheduler: Broadcasting large task binary with size 1174.1 KiB\n",
      "23/07/11 22:38:59 WARN DAGScheduler: Broadcasting large task binary with size 1092.3 KiB\n",
      "23/07/11 22:38:59 WARN DAGScheduler: Broadcasting large task binary with size 1174.1 KiB\n",
      "23/07/11 22:39:00 WARN DAGScheduler: Broadcasting large task binary with size 1091.7 KiB\n",
      "23/07/11 22:39:01 WARN DAGScheduler: Broadcasting large task binary with size 1303.6 KiB\n",
      "23/07/11 22:39:01 WARN DAGScheduler: Broadcasting large task binary with size 1586.6 KiB\n",
      "23/07/11 22:39:01 WARN DAGScheduler: Broadcasting large task binary with size 1247.4 KiB\n",
      "23/07/11 22:39:02 WARN DAGScheduler: Broadcasting large task binary with size 1060.8 KiB\n",
      "23/07/11 22:39:02 WARN DAGScheduler: Broadcasting large task binary with size 1320.8 KiB\n",
      "23/07/11 22:39:03 WARN DAGScheduler: Broadcasting large task binary with size 1738.0 KiB\n",
      "23/07/11 22:39:03 WARN DAGScheduler: Broadcasting large task binary with size 2.3 MiB\n",
      "23/07/11 22:39:03 WARN DAGScheduler: Broadcasting large task binary with size 3.0 MiB\n",
      "23/07/11 22:39:04 WARN DAGScheduler: Broadcasting large task binary with size 2.1 MiB\n",
      "23/07/11 22:39:05 WARN DAGScheduler: Broadcasting large task binary with size 1092.3 KiB\n",
      "23/07/11 22:39:05 WARN DAGScheduler: Broadcasting large task binary with size 1414.9 KiB\n",
      "23/07/11 22:39:05 WARN DAGScheduler: Broadcasting large task binary with size 1945.8 KiB\n",
      "23/07/11 22:39:06 WARN DAGScheduler: Broadcasting large task binary with size 2.7 MiB\n",
      "23/07/11 22:39:06 WARN DAGScheduler: Broadcasting large task binary with size 3.9 MiB\n",
      "23/07/11 22:39:07 WARN DAGScheduler: Broadcasting large task binary with size 5.4 MiB\n",
      "23/07/11 22:39:08 WARN DAGScheduler: Broadcasting large task binary with size 3.6 MiB\n",
      "23/07/11 22:39:09 WARN DAGScheduler: Broadcasting large task binary with size 1091.7 KiB\n",
      "23/07/11 22:39:09 WARN DAGScheduler: Broadcasting large task binary with size 1303.7 KiB\n",
      "23/07/11 22:39:09 WARN DAGScheduler: Broadcasting large task binary with size 1586.6 KiB\n",
      "23/07/11 22:39:10 WARN DAGScheduler: Broadcasting large task binary with size 1247.4 KiB\n",
      "23/07/11 22:39:11 WARN DAGScheduler: Broadcasting large task binary with size 1060.8 KiB\n",
      "23/07/11 22:39:11 WARN DAGScheduler: Broadcasting large task binary with size 1320.8 KiB\n",
      "23/07/11 22:39:11 WARN DAGScheduler: Broadcasting large task binary with size 1738.0 KiB\n",
      "23/07/11 22:39:11 WARN DAGScheduler: Broadcasting large task binary with size 2.3 MiB\n",
      "23/07/11 22:39:12 WARN DAGScheduler: Broadcasting large task binary with size 3.0 MiB\n",
      "23/07/11 22:39:12 WARN DAGScheduler: Broadcasting large task binary with size 2.1 MiB\n",
      "23/07/11 22:39:14 WARN DAGScheduler: Broadcasting large task binary with size 1092.3 KiB\n",
      "23/07/11 22:39:14 WARN DAGScheduler: Broadcasting large task binary with size 1414.9 KiB\n",
      "23/07/11 22:39:14 WARN DAGScheduler: Broadcasting large task binary with size 1945.8 KiB\n",
      "23/07/11 22:39:14 WARN DAGScheduler: Broadcasting large task binary with size 2.7 MiB\n",
      "23/07/11 22:39:15 WARN DAGScheduler: Broadcasting large task binary with size 3.9 MiB\n",
      "23/07/11 22:39:16 WARN DAGScheduler: Broadcasting large task binary with size 5.4 MiB\n",
      "23/07/11 22:39:17 WARN DAGScheduler: Broadcasting large task binary with size 3.6 MiB\n",
      "23/07/11 22:39:28 WARN DAGScheduler: Broadcasting large task binary with size 1104.9 KiB\n",
      "23/07/11 22:39:29 WARN DAGScheduler: Broadcasting large task binary with size 1177.6 KiB\n",
      "23/07/11 22:39:32 WARN DAGScheduler: Broadcasting large task binary with size 1104.9 KiB\n",
      "23/07/11 22:39:33 WARN DAGScheduler: Broadcasting large task binary with size 1177.6 KiB\n",
      "23/07/11 22:39:34 WARN DAGScheduler: Broadcasting large task binary with size 1169.0 KiB\n",
      "23/07/11 22:39:34 WARN DAGScheduler: Broadcasting large task binary with size 1427.8 KiB\n",
      "23/07/11 22:39:34 WARN DAGScheduler: Broadcasting large task binary with size 1753.0 KiB\n",
      "23/07/11 22:39:35 WARN DAGScheduler: Broadcasting large task binary with size 1331.5 KiB\n",
      "23/07/11 22:39:36 WARN DAGScheduler: Broadcasting large task binary with size 1066.0 KiB\n",
      "23/07/11 22:39:36 WARN DAGScheduler: Broadcasting large task binary with size 1337.4 KiB\n",
      "23/07/11 22:39:36 WARN DAGScheduler: Broadcasting large task binary with size 1764.8 KiB\n",
      "23/07/11 22:39:36 WARN DAGScheduler: Broadcasting large task binary with size 2.3 MiB\n",
      "23/07/11 22:39:37 WARN DAGScheduler: Broadcasting large task binary with size 3.0 MiB\n",
      "23/07/11 22:39:37 WARN DAGScheduler: Broadcasting large task binary with size 2.1 MiB\n",
      "23/07/11 22:39:39 WARN DAGScheduler: Broadcasting large task binary with size 1104.9 KiB\n",
      "23/07/11 22:39:39 WARN DAGScheduler: Broadcasting large task binary with size 1444.0 KiB\n",
      "23/07/11 22:39:39 WARN DAGScheduler: Broadcasting large task binary with size 2000.4 KiB\n",
      "23/07/11 22:39:40 WARN DAGScheduler: Broadcasting large task binary with size 2.8 MiB\n",
      "23/07/11 22:39:40 WARN DAGScheduler: Broadcasting large task binary with size 4.0 MiB\n",
      "23/07/11 22:39:41 WARN DAGScheduler: Broadcasting large task binary with size 5.5 MiB\n",
      "23/07/11 22:39:42 WARN DAGScheduler: Broadcasting large task binary with size 3.6 MiB\n",
      "23/07/11 22:39:44 WARN DAGScheduler: Broadcasting large task binary with size 1169.0 KiB\n",
      "23/07/11 22:39:44 WARN DAGScheduler: Broadcasting large task binary with size 1427.8 KiB\n",
      "23/07/11 22:39:44 WARN DAGScheduler: Broadcasting large task binary with size 1753.0 KiB\n",
      "23/07/11 22:39:44 WARN DAGScheduler: Broadcasting large task binary with size 1331.5 KiB\n",
      "23/07/11 22:39:45 WARN DAGScheduler: Broadcasting large task binary with size 1066.0 KiB\n",
      "23/07/11 22:39:45 WARN DAGScheduler: Broadcasting large task binary with size 1337.4 KiB\n",
      "23/07/11 22:39:46 WARN DAGScheduler: Broadcasting large task binary with size 1764.8 KiB\n",
      "23/07/11 22:39:46 WARN DAGScheduler: Broadcasting large task binary with size 2.3 MiB\n",
      "23/07/11 22:39:47 WARN DAGScheduler: Broadcasting large task binary with size 3.0 MiB\n",
      "23/07/11 22:39:47 WARN DAGScheduler: Broadcasting large task binary with size 2.1 MiB\n",
      "23/07/11 22:39:48 WARN DAGScheduler: Broadcasting large task binary with size 1104.9 KiB\n",
      "23/07/11 22:39:48 WARN DAGScheduler: Broadcasting large task binary with size 1444.0 KiB\n",
      "23/07/11 22:39:49 WARN DAGScheduler: Broadcasting large task binary with size 2000.4 KiB\n",
      "23/07/11 22:39:49 WARN DAGScheduler: Broadcasting large task binary with size 2.8 MiB\n",
      "23/07/11 22:39:50 WARN DAGScheduler: Broadcasting large task binary with size 4.0 MiB\n",
      "23/07/11 22:39:51 WARN DAGScheduler: Broadcasting large task binary with size 5.5 MiB\n",
      "23/07/11 22:39:52 WARN DAGScheduler: Broadcasting large task binary with size 3.6 MiB\n",
      "23/07/11 22:39:55 WARN DAGScheduler: Broadcasting large task binary with size 1304.8 KiB\n",
      "23/07/11 22:39:55 WARN DAGScheduler: Broadcasting large task binary with size 1876.9 KiB\n",
      "23/07/11 22:39:56 WARN DAGScheduler: Broadcasting large task binary with size 2.8 MiB\n",
      "23/07/11 22:39:56 WARN DAGScheduler: Broadcasting large task binary with size 4.1 MiB\n",
      "23/07/11 22:39:57 WARN DAGScheduler: Broadcasting large task binary with size 5.9 MiB\n",
      "23/07/11 22:40:00 WARN DAGScheduler: Broadcasting large task binary with size 3.8 MiB\n",
      "23/07/11 22:40:02 WARN DAGScheduler: Broadcasting large task binary with size 3.8 MiB\n",
      "23/07/11 22:40:03 WARN DAGScheduler: Broadcasting large task binary with size 3.8 MiB\n",
      "2023/07/11 22:40:25 WARNING mlflow.utils.environment: Encountered an unexpected error while inferring pip requirements (model URI: /tmp/tmpm86dr5br/model, flavor: spark), fall back to return ['pyspark==3.4.1']. Set logging level to DEBUG to see the full traceback.\n"
     ]
    }
   ],
   "source": [
    "with mlflow.start_run(run_name='random_forest_with_grid'):\n",
    "    # define model and set a tag\n",
    "    randomForest_model = RandomForestClassifier(featuresCol='features', labelCol='puntaje_saber11')\n",
    "    mlflow.set_tag(\"model_name\", str(randomForest_model.__class__.__name__))\n",
    "\n",
    "    # build parameter grid\n",
    "    paramGridTree = ParamGridBuilder()\\\n",
    "        .addGrid(randomForest_model.maxDepth, [3, 5, 10])\\\n",
    "        .addGrid(randomForest_model.maxBins, [16, 32])\\\n",
    "        .addGrid(randomForest_model.numTrees, [20, 50, 100])\\\n",
    "        .build()\n",
    "    \n",
    "    # create an evaluator\n",
    "    evaluator = BinaryClassificationEvaluator(rawPredictionCol='probability', labelCol='puntaje_saber11')\n",
    "\n",
    "    # create cross validator\n",
    "    crossvalRandomForest = CrossValidator(estimator=randomForest_model,\n",
    "                                  estimatorParamMaps=paramGridTree,\n",
    "                                  evaluator=evaluator,\n",
    "                                  numFolds=3)\n",
    "    \n",
    "    # define pipeline with stages and train it\n",
    "    pipeline = Pipeline(stages=[features_pipeline, vector, crossvalRandomForest])\n",
    "    model = pipeline.fit(trainingData)\n",
    "\n",
    "    # evaluate and log metrics and model\n",
    "    trainPredictions = model.transform(trainingData)\n",
    "    trainAUC = evaluator.evaluate(trainPredictions)\n",
    "    testPredictions = model.transform(testingData)\n",
    "    testAUC = evaluator.evaluate(testPredictions)\n",
    "    best_model = model.stages[-1].bestModel\n",
    "    precision, recall, accuracy =  create_metrics(trainPredictions)\n",
    "    mlflow.log_params(params={\"maxDepth\": best_model.getMaxDepth(), \"maxBins\": best_model.getMaxBins(), \"numTrees\": best_model.getNumTrees})\n",
    "    mlflow.log_metrics({\"train AUC\": trainAUC, \"test AUC\": testAUC, \"training accuracy\": accuracy,\n",
    "                    \"training weighted precision\": precision, \n",
    "                    \"training weighted recall\": recall})\n",
    "    new_pipeline =  Pipeline(stages=[features_pipeline, vector, best_model])\n",
    "    new_model = new_pipeline.fit(trainingData)\n",
    "    mlflow.spark.log_model(new_model, 'RandomForest_Spark')"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Model registry"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "from mlflow.tracking import MlflowClient\n",
    "ML_FLOW_TRACKING_URI = 'sqlite:///mlflow.db'"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Extract run ids with higher accuracy"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[<Experiment: artifact_location='/home/josealcocer27/git-projects/PySpark-and-mlflow/mlruns/1', creation_time=1689132767191, experiment_id='1', last_update_time=1689132767191, lifecycle_stage='active', name='saber11', tags={}>,\n",
       " <Experiment: artifact_location='/home/josealcocer27/git-projects/PySpark-and-mlflow/mlruns/0', creation_time=1689132767171, experiment_id='0', last_update_time=1689132767171, lifecycle_stage='active', name='Default', tags={}>]"
      ]
     },
     "execution_count": 24,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "client = MlflowClient(tracking_uri=ML_FLOW_TRACKING_URI)\n",
    "client.search_experiments()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [],
   "source": [
    "from mlflow.entities import ViewType\n",
    "\n",
    "runs = client.search_runs(\n",
    "    experiment_ids='1',\n",
    "    filter_string='metrics.\"train AUC\" > 0.7',\n",
    "    run_view_type=ViewType.ACTIVE_ONLY,\n",
    "    max_results=3,\n",
    "    order_by=['metrics.\"train AUC\" DESC']\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 77,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "run id: 63e932b8215b4fdbbfd4ea476314460d, train AUC: 0.8348,  artifact: RandomForest_Spark\n",
      "run id: 4dd12d7a3fcc467da88109ed96784127, train AUC: 0.7353,  artifact: DecisionTree_Spark\n"
     ]
    }
   ],
   "source": [
    "for run in runs:\n",
    "    artifact = client.list_artifacts(run_id=run.info.run_id)[0].path\n",
    "    print(f\"run id: {run.info.run_id}, train AUC: {run.data.metrics['train AUC']:.4f},  artifact: {artifact}\")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Registry Random Forest"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 88,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Successfully registered model 'saber11_model'.\n",
      "2023/08/03 22:33:39 INFO mlflow.tracking._model_registry.client: Waiting up to 300 seconds for model version to finish creation. Model name: saber11_model, version 1\n",
      "Created version '1' of model 'saber11_model'.\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "<ModelVersion: aliases=[], creation_timestamp=1691120019347, current_stage='None', description=None, last_updated_timestamp=1691120019347, name='saber11_model', run_id='63e932b8215b4fdbbfd4ea476314460d', run_link=None, source='/home/josealcocer27/git-projects/PySpark-and-mlflow/mlruns/1/63e932b8215b4fdbbfd4ea476314460d/artifacts/RandomForest_Spark', status='READY', status_message=None, tags={}, user_id=None, version=1>"
      ]
     },
     "execution_count": 88,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "mlflow.set_tracking_uri(ML_FLOW_TRACKING_URI)\n",
    "\n",
    "run_id = '63e932b8215b4fdbbfd4ea476314460d'\n",
    "artifact = 'RandomForest_Spark'\n",
    "model_uri = f'runs:/{run_id}/{artifact}'\n",
    "mlflow.register_model(model_uri=model_uri, name = 'saber11_model')"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Registry Decision Tree"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 89,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Registered model 'saber11_model' already exists. Creating a new version of this model...\n",
      "2023/08/03 22:33:53 INFO mlflow.tracking._model_registry.client: Waiting up to 300 seconds for model version to finish creation. Model name: saber11_model, version 2\n",
      "Created version '2' of model 'saber11_model'.\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "<ModelVersion: aliases=[], creation_timestamp=1691120033048, current_stage='None', description=None, last_updated_timestamp=1691120033048, name='saber11_model', run_id='4dd12d7a3fcc467da88109ed96784127', run_link=None, source='/home/josealcocer27/git-projects/PySpark-and-mlflow/mlruns/1/4dd12d7a3fcc467da88109ed96784127/artifacts/DecisionTree_Spark', status='READY', status_message=None, tags={}, user_id=None, version=2>"
      ]
     },
     "execution_count": 89,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "run_id = '4dd12d7a3fcc467da88109ed96784127'\n",
    "artifact = 'DecisionTree_Spark'\n",
    "model_uri = f'runs:/{run_id}/{artifact}'\n",
    "mlflow.register_model(model_uri=model_uri, name = 'saber11_model')"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Changing stage to production"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 92,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<ModelVersion: aliases=[], creation_timestamp=1691120019347, current_stage='Production', description=None, last_updated_timestamp=1691120113767, name='saber11_model', run_id='63e932b8215b4fdbbfd4ea476314460d', run_link=None, source='/home/josealcocer27/git-projects/PySpark-and-mlflow/mlruns/1/63e932b8215b4fdbbfd4ea476314460d/artifacts/RandomForest_Spark', status='READY', status_message=None, tags={}, user_id=None, version=1>"
      ]
     },
     "execution_count": 92,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "model_name = 'saber11_model'\n",
    "model_version = 1\n",
    "new_stage = 'Production'\n",
    "client.transition_model_version_stage(\n",
    "    name=model_name,\n",
    "    version=model_version,\n",
    "    stage=new_stage,\n",
    "    archive_existing_versions=False\n",
    ")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Checking versions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 93,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Version: 1, stage: Production\n",
      "Version: 2, stage: None\n"
     ]
    }
   ],
   "source": [
    "latest_versions = client.get_latest_versions(name = model_name, stages=None)\n",
    "\n",
    "for version in latest_versions:\n",
    "    print(f\"Version: {version.version}, stage: {version.current_stage}\")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Make predictions in production"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 103,
   "metadata": {},
   "outputs": [],
   "source": [
    "def make_prediction(name, stage, X):\n",
    "    model = mlflow.spark.load_model(f\"models:/{name}/{stage}\")\n",
    "    data = model.transform(X)\n",
    "    return data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 106,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2023/08/03 22:56:05 INFO mlflow.spark: 'models:/saber11_model/Production' resolved as '/home/josealcocer27/git-projects/PySpark-and-mlflow/mlruns/1/63e932b8215b4fdbbfd4ea476314460d/artifacts/RandomForest_Spark'\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2023/08/03 22:56:05 INFO mlflow.spark: URI 'models:/saber11_model/Production/sparkml' does not point to the current DFS.\n",
      "2023/08/03 22:56:05 INFO mlflow.spark: File 'models:/saber11_model/Production/sparkml' not found on DFS. Will attempt to upload the file.\n",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "preds = make_prediction(name='saber11_model', stage='Production', X = testingData)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 118,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "23/08/03 22:59:47 WARN DAGScheduler: Broadcasting large task binary with size 4.1 MiB\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "0.5842636895268474"
      ]
     },
     "execution_count": 118,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from sklearn.metrics import accuracy_score\n",
    "\n",
    "aux_preds = preds.select('puntaje_saber11', 'prediction').toPandas()\n",
    "y_true = aux_preds['puntaje_saber11'].values\n",
    "y_predict = aux_preds['prediction'].values\n",
    "accuracy_score(y_true, y_predict)"
   ]
  }
 ],
 "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"
  },
  "orig_nbformat": 4
 },
 "nbformat": 4,
 "nbformat_minor": 2
}