Parte 2. S05. Mineria de datos y Machine Learning con SparkR.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": [
    "# Aplicando técnicas de Machine Learning en SparkR\n",
    "\n",
    "![Spark+R](https://sites.google.com/site/manuparra/home/SparkRlogo.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Como siempre para todos nuestros scripts con SparkR, cargamos la biblioteca, y creamos una nueva sesión de SparkR."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    ".libPaths(c(file.path(Sys.getenv(\"SPARK_HOME\"),\"R/lib/\"),.libPaths()))\n",
    "library(SparkR)\n",
    "sparkR.session(appName=\"EntornoInicio\", master = \"local[*]\", sparkConfig = list(spark.driver.memory = \"1g\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<HR>\n",
    "La biblioteca de ``SparkR`` actualmente soporta los siguientes algoritmos de aprendizaje automático : \n",
    "\n",
    "- modelo lineal generalizado, \n",
    "- modelo de regresión de supervivencia con tiempo de fallo acelerado (AFT), \n",
    "- modelo Bayes Naive y \n",
    "- modelo KMeans. \n",
    "\n",
    "SparkR utiliza MLlib para entrenar el modelo. Por tanto se puede analizar el resumen del modelo ajustado, predecir y   hacer predicciones sobre nuevos datos y escribir/leer el modelo para guardar / cargar los modelos ajustados. \n",
    "\n",
    "Además de ello, al igual que ocurre cuando usamos cualquier funcion en R, SparkR soporta el uso de fómulas, lo cual mejora bastante la adopción de SparkR para análisis de datos másivos.  SparkR soporta un subconjunto de los operadores de fórmula R disponibles para el ajuste del modelo, incluyendo '~', '.', ':', '+' y '-'."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "####  Funciones enmascaradas\n",
    "\n",
    "Dado que parte de SparkR está modelado en el paquete dplyr, ciertas funciones de SparkR comparten los mismos nombres con los de dplyr. Dependiendo del orden de carga de los dos paquetes, algunas funciones del paquete cargado primero son enmascaradas por las del paquete cargado después. \n",
    "\n",
    "``cov in package:stats``\n",
    "\n",
    "``filter in package:stats``\n",
    "\n",
    "``sample in package:base``\n",
    "\n",
    "\n",
    "Por tanto hay siempre que usar el paquete que queramos usar al final de la importación de las bibliotecas para que se haga efectiva la función que queremos para SparkR.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Algoritmos\n",
    "\n",
    "El paquete SparkR soporta las siguientes funcionalidades de Machine Learning y Data mining"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Generalized Linear Model"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Usamos la función de R"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "\n",
    "gaussianDF <- iris\n",
    "gaussianTestDF <- iris\n",
    "gaussianGLM <- glm(data = gaussianDF, Sepal.Length ~ Sepal.Width + Species, family = \"gaussian\")\n",
    "\n",
    "summary(gaussianGLM)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Usamos la función para glm de SparkR"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "\n",
    "irisDF <- suppressWarnings(createDataFrame(iris))\n",
    "# Fit a generalized linear model of family \"gaussian\" with spark.glm\n",
    "gaussianDF <- irisDF\n",
    "gaussianTestDF <- irisDF\n",
    "gaussianGLM <- spark.glm(gaussianDF, Sepal_Length ~ Sepal_Width + Species, family = \"gaussian\")\n",
    "\n",
    "# Model summary\n",
    "summary(gaussianGLM)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**¿Qué diferencias hay entre las dos?**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Calculamos el modelo"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Calculamos la predicción\n",
    "gaussianPredictions <- predict(gaussianGLM, gaussianTestDF)\n",
    "# Mostramos las predicciones\n",
    "showDF(gaussianPredictions)\n",
    "\n",
    "# Usamos la función de R de glm con la familia gaussian\n",
    "gaussianGLM2 <- glm(Sepal_Length ~ Sepal_Width + Species, gaussianDF, family = \"gaussian\")\n",
    "summary(gaussianGLM2)\n",
    "\n",
    "# Ahora usamos la función de glm de spark para la familia binomial.\n",
    "binomialDF <- filter(irisDF, irisDF$Species != \"setosa\")\n",
    "binomialTestDF <- binomialDF\n",
    "binomialGLM <- spark.glm(binomialDF, Species ~ Sepal_Length + Sepal_Width, family = \"binomial\")\n",
    "\n",
    "# Imprimimos el modelo\n",
    "summary(binomialGLM)\n",
    "\n",
    "# Obtenemos la predicción\n",
    "binomialPredictions <- predict(binomialGLM, binomialTestDF)\n",
    "showDF(binomialPredictions)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Accelerated Failure Time (AFT) Survival Regression Model"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "``spark.survreg()`` ajusta a un modelo de regresión AFT (accelerated failure time) sobre un SparkDataFrame. Para esta función no se permite el uso del operador . en la formula."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Use the ovarian dataset available in R survival package\n",
    "library(survival)\n",
    "\n",
    "ovarianDF <- suppressWarnings(createDataFrame(ovarian))\n",
    "\n",
    "head(ovarianDF)\n",
    "\n",
    "aftDF <- ovarianDF\n",
    "aftTestDF <- ovarianDF\n",
    "\n",
    "#Aplicamos la función\n",
    "aftModel <- spark.survreg(aftDF, Surv(futime, fustat) ~ ecog_ps + rx)\n",
    "\n",
    "# Model summary\n",
    "summary(aftModel)\n",
    "\n",
    "# Prediction\n",
    "aftPredictions <- predict(aftModel, aftTestDF)\n",
    "showDF(aftPredictions)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### K-Means"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "\n",
    "library(ggplot2)\n",
    "ggplot(iris, aes(Petal.Length, Petal.Width, color = Species)) + geom_point()\n",
    "\n",
    "set.seed(20)\n",
    "irisCluster <- kmeans(iris[, 3:4], 3, nstart = 20)\n",
    "irisCluster\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Ajustamos un modelo k-medias. \n",
    "\n",
    "irisDF <- suppressWarnings(createDataFrame(iris))\n",
    "kmeansDF <- irisDF\n",
    "kmeansTestDF <- irisDF\n",
    "kmeansModel <- spark.kmeans(kmeansDF, ~ Sepal_Length + Sepal_Width + Petal_Length + Petal_Width,\n",
    "                            k = 3)\n",
    "\n",
    "# Vemos el resumen\n",
    "summary(kmeansModel)\n",
    "\n",
    "# Vemos los resultados del ajuste\n",
    "showDF(fitted(kmeansModel))\n",
    "\n",
    "# y vemos la predicción\n",
    "kmeansPredictions <- predict(kmeansModel, kmeansTestDF)\n",
    "showDF(kmeansPredictions)\n",
    "\n",
    "# Mostramos la información de los grupos.\n",
    "table(kmeansModel$cluster, iris$Species)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "### Naive Bayes\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "### Aplicamos NAIVE.BAYES \n",
    "\n",
    "titanic <- as.data.frame(Titanic)\n",
    "titanicDF <- createDataFrame(titanic[titanic$Freq > 0, -5])\n",
    "nbDF <- titanicDF\n",
    "nbTestDF <- titanicDF\n",
    "nbModel <- spark.naiveBayes(nbDF, Survived ~ Class + Sex + Age)\n",
    "\n",
    "# Resumen del modelo\n",
    "summary(nbModel)\n",
    "\n",
    "# Predicción\n",
    "nbPredictions <- predict(nbModel, nbTestDF)\n",
    "showDF(nbPredictions)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "### Creación de Conjuntos de entrenamiento y prueba\n",
    "\n",
    "Existen varias formas de hacer los conjuntos de prueba y test. Se pueden usar las funciones de muestreo (sample) que trabajan sobre los SparkDataFrames."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "train_df <- sample(df_training, withReplacement=FALSE, fraction=0.85, seed=42)\n",
    "test_df <- except(df_training, train_df)\n",
    "\n",
    "count(train_df)\n",
    "count(test_df)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Persistencia de los MODELOS de ML.\n",
    "\n",
    "Si necesitamos almacenar el modelo que hemos ajustado podemo hacerlo mediante el uso de la funcion ``write.ml``. \n",
    "Al igual que luego podemos recuperarlo con ``read.ml``."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "modelPath <- tempfile(pattern = \"ml\", fileext = \".tmp\")\n",
    "write.ml(gaussianGLM, modelPath)\n",
    "gaussianGLM2 <- read.ml(modelPath)"
   ]
  },
  {
   "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": false
   },
   "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
}