Parte 2. S06. Mineria de datos y Machine Learning con sparklyr.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<span style='font-size:2.4em'>Procesamiento masivo de datos con SparkR</span>\n",
    "\n",
    "<span style='font-size:1.5em'>VIII Jornadas de usuarios de R. Albacete, Castilla-La Mancha, 17 y 18 de noviembre de 2016</span>\n",
    "\n",
    "Taller impartido por: <span style='font-size:1.2em'>Manuel Jesús Parra Royón</span>\n",
    "\n",
    "\n",
    "![Alt](https://sites.google.com/site/manuparra/home/logoparty.png)\n",
    "\n",
    "<HR>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Machine Learning con sparklyr — Interfaz R para Apache Spark y la MLLib"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "** Biblioteca de Machine Learning Spark (MLlib) con la Interfaz sparklyr **\n",
    "\n",
    "*Carácteristicas fundamentales:*\n",
    "\n",
    "La bilbioteca ```sparklyr``` proporciona enlaces a la biblioteca de ML distribuida de Spark. En particular, sparklyr le permite acceder a las rutinas de ML proporcionadas por el paquete ``spark.ml`` de Spark. Además junto con la interfaz ``dplyr`` de ``sparklyr``, puede crear y afinar fácilmente los flujos de trabajo de ML en Spark, orquestados enteramente dentro de ``R``.\n",
    "\n",
    "Sparklyr proporciona tres familias de funciones que puede utilizar con el aprendizaje de máquina Spark:\n",
    "\n",
    "* Algoritmos de aprendizaje automático para el análisis de datos (funciones  ``ml_*``)\n",
    "* Transformadores de características para manipular características individuales (funciones ``ft_*``)\n",
    "* Funciones para manipular Spark DataFrames (funciones ``sdf_*``)\n",
    "\n",
    "El flujo de trabajo para el análisis de datos  con ``sparklyr`` podría estar compuesto de las siguientes etapas:\n",
    "\n",
    "- Realizar consultas SQL a través de la interfaz sparklyr dplyr,\n",
    "- Utilizar la familia de funciones ``sdf_`` * y ``ft_`` * para generar nuevas columnas o particionar su conjunto de datos,\n",
    "- Elegir un algoritmo de aprendizaje automático apropiado de la familia de funciones ``ml_ *`` para modelar los datos,\n",
    "- Inspeccionar la calidad del ajuste de su modelo y usarlo para hacer predicciones con nuevos datos,\n",
    "- Recopilar los resultados para la visualización y análisis posterior en R\n",
    "\n",
    "<HR size=1>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "En esta sección vamos a trabajar con las herramientas que proporciona la biblioteca ``sparklyr``.\n",
    "\n",
    "Como vamos a ver, la funcionalidad es muy similar a la de la biblioteca SparkR, aunque cambia el nombre de las funciones y varios extras más."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Inicialización del entorno"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<span style=\"background-color:red;color:white\">&nbsp; &nbsp; Es necesario reiniciar Spark para poder trabajar con esta sesión de sparklyr &nbsp; &nbsp; <BR>&nbsp; &nbsp; Para ello, ve a la Máquina Virtual para el proceso de Spark, para luego volver a cargarlo &nbsp; &nbsp;</span> \n",
    "\n",
    "Ahora para conectar con Spark y abrir una sesión usaremos la siguiente sintaxis (simiar a la del paquete SparkR aunque  con ligeras diferencias):"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Usamos la libreria sparklyr y dplyr.\n",
    "# Ajustar el nivel de visualización de errores !\n",
    "options(warn=0)\n",
    "\n",
    "# Incluimos la bilbioteca de sparklyr\n",
    "library(sparklyr)\n",
    "# Usamos la biblioteca para el manejo de los datos.\n",
    "library(dplyr)\n",
    "\n",
    "# Abrimos la conexión. Importante indicar la versión de Spark que tenemos instalada. En nuestro caso tenemos la 2.0.1\n",
    "sc <- spark_connect(master = \"local\", version = \"2.0.1\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Características"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "La biblioteca sparklyr tiene asociado un paquete que hace de complemento ideal para la manipulación de datos masivos. Este paquete es ``dplyr``  un paquete en R para trabajar con datos estructurados dentro y fuera de R. ``dplyr`` hace la manipulación de datos muy sencilla para los usuarios de R, además ofrece interfaces consistentes y con un buen rendimiento. La librería tiene las siguientes funcionalidades básicas: \n",
    "\n",
    "* Seleccion, filtrado y agregación.\n",
    "* Funciones windows (para muestreo).\n",
    "* Funciones de JOIN para Dataframes.\n",
    "* Funciones Collect para transformar datos de Spark a R.\n",
    "* ...\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Lectura y escritura de datos con ``sparklyr``"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "** Para la lectura y escritura de datos tenemos las siguientes funciones:**\n",
    "\n",
    "- ``spark_read_csv``: Lee un CSV y el resultado lo hace compatible con las funciones de ``dplyr``.\n",
    "- ``spark_read_json``: Lee un fichero JSON y el resultado es compatible con la interfaz de ``dplyr``.\n",
    "- ``spark_read_parquet``: Lee un fichero PARQUET.\n",
    "\n",
    "Además del formato de los datos, Spark soporta la lectura de datos desde una variedad de fuentes de datos. Estos incluyen, almacenamiento en  HDFS (hdfs:// protocol), Amazon S3 (s3n:// protocol), o ficheros locales disponibles en en los nodos (file:// protocol)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "** Para la escritura de DataFrames existen las mismas funcione según el tipo de fuente de datos:**\n",
    "\n",
    "- ``spark_write_csv``: Escribe a CSV y recibe una fuente de datos compatible con ``dplyr``.\n",
    "- ``spark_write_json``: Escribe a  JSON.\n",
    "- ``spark_write_parquet``: Escribe a parquet desde cualquier fuente compatible con ``dplyr``."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Lectura de un fichero de datos CSV\n",
    "\n",
    "tttm <- spark_read_csv(sc, \n",
    "                       name=\"tttm\", \n",
    "                       path=\"/root/TallerSparkR/datasets/databig/ECBDL14_10tst.data\", \n",
    "                       delimiter = \",\", \n",
    "                       header=TRUE,\n",
    "                       overwrite = TRUE)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "¿Carga los datos rápido?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "count(tttm)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "La escritura de datos es sencilla y simplemente requiere la función concreta para almacenar los datos.\n",
    "\n",
    "El valor del parámetros ``path`` puede ser de diferente origen de datos:\n",
    "\n",
    "    * HDFS (``path=\"hdfs://....\"``)\n",
    "    * AmazonS3 (``path=\"s3://...\"``)\n",
    "    * Local (``path=\"...\"``)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Escritura de un fichero de datos CSV\n",
    "\n",
    "spark_write_csv(tttm, \n",
    "                path=\"/root/TallerSparkR/datasets/databig/ECBDL14_10tst_saved.data\", \n",
    "                delimiter = \",\", \n",
    "                header=TRUE)\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Para los demás formatos, se usa la función correspondiente, teniendo en cuenta que la entrada de la función de escritura, siempre tiene que ser compatible con ``dplyr``"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#spark_write_json(tttm, .....\n",
    "\n",
    "#spark_write_parquet(tttm, ....."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Filtrado, selección, agrupación."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Las funciones son comandos dplyr para manipular datos... for manipulating data. When connected to a Spark DataFrame, dplyr translates the commands into Spark SQL statements. Remote data sources use exactly the same five verbs as local data sources. Here are the five verbs with their corresponding SQL commands:\n",
    "\n",
    "* select ~ SELECT\n",
    "* filter ~ WHERE\n",
    "* arrange ~ ORDER\n",
    "* summarise ~ aggregators: sum, min, sd, etc.\n",
    "* mutate ~ operators: +, *, log, etc.\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "head(tttm)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Los resultados y las tablas, ya no son tan \"bonitas\" como con la biblioteca SparkR."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Seleccionamos las columnas que queremos con select\n",
    "select(tttm, f5,f10,class)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Filtrado de instancias"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Hago un filtrado de datos tal que la columna f5 sea mayor de 0 y la clase sea 1\n",
    "res_filtered<-filter(tttm, f5 > 0 & class==1)\n",
    "\n",
    "# Contamos los registros.\n",
    "count(res_filtered)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### ¿Cómo es el dataset: Balanceado o no Balanceado con respecto a la variable de clase ``class`` ?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Agrupamos el dataset por clase y luego contamos los registros.\n",
    "\n",
    "# En SQL sería select count(class) ...group by class\n",
    "num_regs <- as.integer(collect(count(tttm)))\n",
    "\n",
    "# Mostramos el número de registros\n",
    "print(num_regs)\n",
    "\n",
    "# Agrupamos por clase y contamos el numero de elementos de cada clase, ademñas \n",
    "# añadimos una columna que calcula el porcentaje que supone cada clase del total\n",
    "summarize( group_by(tttm,class), count = n(), percent= n()/num_regs *100.0)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "La clase con valor *1*, tiene sólo un *1.67%* de instancias del total, por tanto, es un dataset NO balanceado. Interesante !."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Preprocesado de dataframes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "library(magrittr)\n",
    "\n",
    "\n",
    "# También podemos usar magittr para hacer los mismo de un modo más claro. El ejemplo de arriba y este son lo mismo.\n",
    "tttm %>% \n",
    "    group_by(class) %>%\n",
    "    summarize(count = n(), percent= n()/num_regs *100.0 )\n",
    "\n",
    "tttm %>% \n",
    "    group_by(class) %>%\n",
    "    summarize(count = n(), maxf1=max(f1),minf1= min(f1))\n",
    "head(tttm)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "\n",
    "# Añadimos una columna con el doble del valor de la columna f6\n",
    "tttm %>%\n",
    "    mutate(f6*2)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Si del anterior ejemplo queremos construir un nuevo dataframe, usamos \n",
    "# select para tomar las columnas que nos interesen.\n",
    "tttm %>%\n",
    "    mutate(f6*2)%>%\n",
    "    select (f1,f8, `f6 * 2` )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# La función arrange permite aplicar funciones de ordenación, ... sobre dataframes.\n",
    "\n",
    "# Si del anterior ejemplo queremos ordenar la columna f8 y filtrar luego por f1 mayor de 0.2\n",
    "tttm %>%\n",
    "    mutate(f6*2) %>%\n",
    "    select (f1,f8, `f6 * 2` ) %>%\n",
    "    arrange (desc(f8)) %>%\n",
    "    filter (f1> 0.2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Muestreo aleatorio de datos.\n",
    "\n",
    "El uso de sampling aleatorio sobre los datos es muy usado para trabajar con dataframes que serán usados para su  tratamiento con Machine Learning o Minería de datos."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Crea una muestra de los datos.  10% de los datos\n",
    "test <- sample_frac(tttm, 0.10)\n",
    "count(test)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Crea una muestra aleatoria de los datos.  80% de los datos\n",
    "train <- sample_frac(tttm, 0.90)\n",
    "count(train)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "También podemos usar:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# sdf_partition(dataset,training = 0.5, test = 0.5, seed = 1099)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Vamos a comprobar como está ahora el porcetaje de balanceo de las clases:"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Comprobamos que más o menos están igual que en proporción que con el conjunto grande cuando hacemos un muestreo."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# También podemos usar magittr para hacer los mismo de un modo más claro. El ejemplo de arriba y este son lo mismo.\n",
    "num_regs <- as.integer(collect(count(test)))\n",
    "\n",
    "print(num_regs)\n",
    "# Calculamos la distribución de clases\n",
    "test %>% \n",
    "    group_by(class) %>%\n",
    "    summarize(count = n(), percent= n()/num_regs *100.0 )\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# También podemos usar magittr para hacer los mismo de un modo más claro. El ejemplo de arriba y este son lo mismo.\n",
    "num_regs <- as.integer(collect(count(train)))\n",
    "\n",
    "# Calculamos la distribución de clases\n",
    "train %>% \n",
    "    group_by(class) %>%\n",
    "    summarize(count = n(), percent= n()/num_regs *100.0 )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Preparación del dataset para Machine Learning\n",
    "\n",
    "Al tener un conjunto de datos grande y desbalanceado, podemos tomar varias alternativas para trabajar con el mismo: \n",
    "\n",
    "* Hacer un sobremuestreo de la clase minoritaria\n",
    "* Hacer un submuestreo de la clase mayoritaria\n",
    "\n",
    "Vamos a trabajar con los algoritmos de ML utilizando submuestreo ``Random Undersampling`` básico, para ello:\n",
    "\n",
    "1. Calculamos el número de instacias de la clase minoritaria.\n",
    "2. Hacemos un muestreo sólo de la clase mayoritaria para igualar en instancias la clase minoritaria.\n",
    "3. Fusionamos ambos muestreos"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Contamos los registros de la clase minoritaria\n",
    "regs_minor <- tttm %>% \n",
    "            filter(class==1) %>% \n",
    "            count %>%\n",
    "            collect %>% \n",
    "            as.integer\n",
    "\n",
    "\n",
    "# Extraemos un sample con un numero similar de instancias que de la clase minoritaria\n",
    "only_class_0 <- tttm %>% \n",
    "        filter(class==0) %>%\n",
    "        sdf_sample(regs_minor, fraction=as.double(regs_minor/as.integer(collect(count(tttm)))))\n",
    "\n",
    "# Extraemos las instancias de la clase minoritaria \n",
    "only_class_1 <- tttm %>% \n",
    "            filter(class==1)\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Contamos los registros de ambos"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "\n",
    "count(only_class_0)\n",
    "# as.integer(collect(count(only_class_0)))\n",
    "# only_class_0 %>% count %>% collect %>% as.integer\n",
    "count(only_class_1)\n",
    "# as.integer(collect(count(only_class_1)))\n",
    "# only_class_1 %>% count %>% collect %>% as.integer\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "¿Por qué no tienen el mismo tamaño?. **Precisión**."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Unimos los dos dataframes con ``rbind``"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Unimos \n",
    "ds_ml <- rbind(only_class_1,only_class_0, name=\"ds_ml\")\n",
    "\n",
    "# Calculamos el tamaño de cada clase.\n",
    "ds_ml %>% \n",
    "        filter(class==0) %>%\n",
    "        count()\n",
    "\n",
    "ds_ml %>% \n",
    "        filter(class==1) %>%\n",
    "        count()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Particionado de datos para conjuntos de Entrenamiento y Prueba (TRAIN & TEST)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Para el particionado usamos una función de sparklyr: ``sdf_partition``"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# La función sdf_partition devuelve el dataframe separado en training y test.\n",
    "# Para acceder a cada dataframe usamos ...$training , ...$test\n",
    "partitions <- sdf_partition(ds_ml,training=0.80,test=0.20)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Contamos el número de registros de cada conjunto:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "count(partitions$test)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "count(partitions$training)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Algoritmos de ML para el análisis de datos\n",
    "\n",
    "Con Spark + R y la biblioteca sparklyr se pueden orquestar algoritmos de ML en un cluster con Spark. \n",
    "\n",
    "Estas funciones de ML, conectan directamente con la API de Spark\n",
    "\n",
    "1. Ajustamos un modelo a nuestro conjunto de entrenamiento\n",
    "2. Evaluamos nuestro rendimiento predictivo sobre el conjunto de test.\n",
    "\n",
    "La biblioteca tiene mayor número de funciones para la minería de datos que la propia de SparkR.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Regresión Lineal\n",
    "\n",
    "En estadística la regresión lineal o ajuste lineal es un modelo matemático usado para aproximar la relación de dependencia entre una variable dependiente Y, las variables independientes Xi y un término aleatorio ε"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# La función para la regresión lineal puede usarse con la sintaxis propia \n",
    "# de la función y tambien con la forma tradicional de R: formulae\n",
    "\n",
    "# Opción 1\n",
    "#model <- partitions$training %>%\n",
    "#    ml_linear_regression(response = \"f1\", features = c(\"f2\",\"f3\"))\n",
    "\n",
    "# Opción 2\n",
    "#model <- partitions$training %>%\n",
    "#    ml_linear_regression(f1~f2+f3)\n",
    "\n",
    "# Opción 3\n",
    "model <- ml_linear_regression(partitions$training,f1~f7)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Vemos la calidad del ajuste ...\n",
    "summary(model)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "predicted <- predict(model, newdata = partitions$test)\n",
    "\n",
    "# extract the true 'strength' values from our test dataset\n",
    "actual <- partitions$test %>%\n",
    "  select(f1) %>%\n",
    "  collect() %>%\n",
    "  `[[`(\"f1\")\n",
    "     "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "\n",
    "# produce a data.frame housing our predicted + actual 'strength' values\n",
    "data <- data.frame(\n",
    "  predicted = predicted,\n",
    "  actual    = actual\n",
    ") "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# plot predicted vs. actual values\n",
    "library(ggplot2)\n",
    "ggplot(data, aes(x = actual, y = predicted)) +\n",
    "  geom_abline(lty = \"dashed\", col = \"red\") +\n",
    "  geom_point(size=0.2) +\n",
    "  theme(plot.title = element_text(hjust = 0.5)) +\n",
    "  coord_fixed(ratio = 4) +\n",
    "  labs(\n",
    "    x = \"Actual F1\",\n",
    "    y = \"Predicted F1\",\n",
    "    title = \"Predicted vs. Actual Feature 1\"\n",
    "  )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Regresión logística\n",
    "\n",
    "En estadística, la regresión logística es un tipo de análisis de regresión utilizado para predecir el resultado de una variable categórica (una variable que puede adoptar un número limitado de categorías) en función de las variables independientes o predictoras. Es útil para modelar la probabilidad de un evento ocurriendo como función de otros factores. El análisis de regresión logística se enmarca en el conjunto de Modelos Lineales Generalizados (GLM por sus siglas en inglés)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Aplica la función de regresión logística\n",
    "ml_log <- partitions$training %>%\n",
    "    ml_logistic_regression(response = \"class\", features = c(\"f1\",\"f2\",\"f3\",\"f4\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "predicted <- predict(model, newdata = partitions$test)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<HR>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### K-Means\n",
    "\n",
    "K-means es un método de agrupamiento, que tiene como objetivo la partición de un conjunto de n observaciones en k grupos en el que cada observación pertenece al grupo cuyo valor medio es más cercano. \n",
    "\n",
    "Forma parte de las técnicas de ``clustering``\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#model <- ml_kmeans(select(iris_tbl,Petal_Width, Petal_Length),centers = 3)\n",
    "\n",
    "model <- iris_tbl %>%\n",
    "  select(Petal_Width, Petal_Length) %>%\n",
    "  ml_kmeans(centers = 3)\n",
    "\n",
    "# Cuidado con collect !!!\n",
    "iris_tbl %>%\n",
    "  select(Petal_Width, Petal_Length) %>%\n",
    "  collect %>%\n",
    "  ggplot(aes(Petal_Length, Petal_Width)) +\n",
    "    geom_point(data = model$centers, aes(Petal_Width, Petal_Length), size = 60, alpha = 0.1) +\n",
    "    geom_point(aes(Petal_Width, Petal_Length), size = 2, alpha = 0.5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Random Forest\n",
    "\n",
    "Random forest también conocido en castellano como '\"Selvas Aleatorias\"' es una combinación de árboles predictores tal que cada árbol depende de los valores de un vector aleatorio probado independientemente y con la misma distribución para cada uno de estos. Es una modificación sustancial de bagging que construye una larga colección de árboles no correlacionados y luego los promedia. Se usa para regresión y para clasificación.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "training_cv <- partitions$training %>% \n",
    "        select(f1,f2,f3,f4,class) %>%\n",
    "        mutate(class1=as.character(class)) %>%\n",
    "        select(f1,f2,f3,f4,class=class1)\n",
    "\n",
    "\n",
    "\n",
    "ml_rf <- ml_random_forest(training_cv,response=\"class\",features=c(\"f1\",\"f2\",\"f3\",\"f4\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "print(fit_model)\n",
    "\n",
    "sdf_predict(fit_model, training_cv)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "#Probamos con el dataset completo. Cuidado !!! Coste en tiempo elevado !\n",
    "ml_rf <- ml_random_forest(tttm,response=\"class\",features=c(\"f1\",\"f2\",\"f3\",\"f4\"),num.trees = 20,type = \"classification\")\n",
    "\n",
    "#sdf_predict(ml_rf, partitions$test)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "ml_rf\n",
    "print(ml_score)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Análisis de componentes principales\n",
    "\n",
    "En estadística, el análisis de componentes principales es una técnica utilizada para reducir la dimensionalidad de un conjunto de datos.\n",
    "\n",
    "Técnicamente, el PCA busca la proyección según la cual los datos queden mejor representados en términos de mínimos cuadrados. Esta convierte un conjunto de observaciones de variables posiblemente correlacionadas en un conjunto de valores de variables sin correlación lineal llamadas componentes principales.\n",
    "\n",
    "El PCA se emplea sobre todo en análisis exploratorio de datos y para construir modelos predictivos."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# PCA\n",
    "pca_model <- ml_pca(partitions$training)\n",
    "print(pca_model)\n",
    "\n",
    "# Calculate explained variance\n",
    "# pca_model$sdev^2 / sum(pca_model$sdev^2) # Sólo para R\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Comparación de modelos "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Hacemos una lista para verificar que modelo nos da mejores resultados.\n",
    "ml_models <- list(\n",
    "  \"Logistic\" = ml_log,\n",
    "#  \"Decision Tree\" = ml_dt,\n",
    "  \"Random Forest\" = ml_rf\n",
    "#  \"Gradient Boosted Trees\" = ml_gbt,\n",
    "#  \"Naive Bayes\" = ml_nb,\n",
    "#  \"Neural Net\" = ml_nn\n",
    ")\n",
    "\n",
    "# Create a function for scoring\n",
    "score_test_data <- function(model, data=partitions$test){\n",
    "  pred <- sdf_predict(model, data)\n",
    "  select(pred, class, prediction)\n",
    "}\n",
    "\n",
    "# Score all the models\n",
    "ml_score <- lapply(ml_models, score_test_data)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<HR>\n",
    "\n",
    "### Cierre de la conexión con SPARK\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "spark_disconnect(sc)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<HR>\n",
    "# Zona de pruebas del NOTEBOOK en SparkR\n",
    "![FooterSparkR](https://sites.google.com/site/manuparra/home/footer_SparkR_v2.png)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "kernelspec": {
   "display_name": "R",
   "language": "R",
   "name": "ir"
  },
  "language_info": {
   "codemirror_mode": "r",
   "file_extension": ".r",
   "mimetype": "text/x-r-source",
   "name": "R",
   "pygments_lexer": "r",
   "version": "3.3.1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}