10 - Spark ML.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![RDD key pair](media/18.sparkml.jpg)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 10 Spark ML: Machine Learning Library\n",
    "\n",
    "Librería de algoritmos paralelos de ML para datos masivos\n",
    "\n",
    "-   Algoritmos clásicos de machine learning: clasificación, regresión, clustering, filtrado colaborativo\n",
    "-   Otros algoritmos: extracción de características, transformación, reducción de dimensionalidad y selección\n",
    "-   Herramientas para construir, evaluar y ajustar pipelines de ML\n",
    "-   Otras utilidades: álgebra lineal, estadística, manejo de datos, etc.\n",
    "\n",
    "\n",
    "Dos paquetes:\n",
    "\n",
    "-   **spark.mllib**: API original, basada en RDDs\n",
    "-   **spark.ml**: API de alto nivel, basada en DataFrames\n",
    "\n",
    "Documentación:\n",
    "[spark.apache.org/docs/latest/mllib-guide.html](http://spark.apache.org/docs/latest/mllib-guide.html)\n",
    "y\n",
    "[spark.apache.org/docs/latest/ml-guide.html](http://spark.apache.org/docs/latest/ml-guide.html)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Requirement already satisfied: pyspark in /home/xergioalex/Envs/tensorflow/lib/python3.5/site-packages\n",
      "Requirement already satisfied: py4j==0.10.7 in /home/xergioalex/Envs/tensorflow/lib/python3.5/site-packages (from pyspark)\n",
      "\u001b[33mYou are using pip version 9.0.1, however version 18.0 is available.\n",
      "You should consider upgrading via the 'pip install --upgrade pip' command.\u001b[0m\n"
     ]
    }
   ],
   "source": [
    "!pip install pyspark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create apache spark context\n",
    "from pyspark import SparkContext\n",
    "sc = SparkContext(master=\"local\", appName=\"Mi app\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Stop apache spark context\n",
    "sc.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Ejemplo\n",
    "\n",
    "Usa el algoritmo de clustering [KMeans](http://spark.apache.org/docs/latest/mllib-clustering.html#k-means) para agrupar datos de vectores dispersos en dos clusters."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[0.  1.2 0. ]\n",
      "[0.  1.1 0. ]\n",
      "[0.9 0.  1. ]\n",
      "[1.  0.  1.1]\n"
     ]
    }
   ],
   "source": [
    "from pyspark.mllib.clustering import KMeans, KMeansModel\n",
    "from pyspark.mllib.linalg import SparseVector\n",
    "import numpy as np\n",
    "\n",
    "# Define un array de 4 vectores dispersos, de 3 elementos cada uno\n",
    "sparse_data = [\n",
    "     SparseVector(3, {1: 1.2}),\n",
    "     SparseVector(3, {1: 1.1}),\n",
    "     SparseVector(3, {0: 0.9, 2: 1.0}),\n",
    "     SparseVector(3, {0: 1.0, 2: 1.1})\n",
    " ]\n",
    "\n",
    "for i in range(4):\n",
    "    print(sparse_data[i].toArray())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Centros de los clusters: [array([0.  , 1.15, 0.  ]), array([0.95, 0.  , 1.05])]\n",
      "Cluster para el nodo 0 = 0\n",
      "Cluster para el nodo 1 = 0\n",
      "Cluster para el nodo 2 = 1\n",
      "Cluster para el nodo 3 = 1\n",
      "\n",
      "Cluster para el nodo (3,[0,1,2],[0.9,1.0,1.0]) = 1\n"
     ]
    }
   ],
   "source": [
    "# Construye el modelo (agrupa los datos en 2 clusters)\n",
    "model = KMeans.train(sc.parallelize(sparse_data), \\\n",
    "                     2, initializationMode=\"k-means||\",\\\n",
    "                     seed=50, initializationSteps=5, \\\n",
    "                     epsilon=1e-4)\n",
    "\n",
    "print(\"Centros de los clusters: {0}\".format(model.clusterCenters))\n",
    "\n",
    "for i in range(4):\n",
    "    print(\"Cluster para el nodo {0} = {1}\"\n",
    "           .format(i, model.predict(sparse_data[i])))\n",
    "\n",
    "punto = SparseVector(3, {0: 0.9, 1:1.0, 2: 1.0})\n",
    "print(\"\\nCluster para el nodo {0} = {1}\".format(punto, model.predict(punto)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "True\n",
      "True\n",
      "True\n",
      "True\n",
      "True\n"
     ]
    }
   ],
   "source": [
    "# Salva el modelo en un directorio temporal\n",
    "import os, tempfile\n",
    "path = tempfile.mkdtemp()\n",
    "model.save(sc, path)\n",
    "\n",
    "# Vuelve a cargar el modelo\n",
    "sameModel = KMeansModel.load(sc, path)\n",
    "\n",
    "for i in range(4):\n",
    "    print(sameModel.predict(sparse_data[i]) == model.predict(sparse_data[i]))\n",
    "\n",
    "print(sameModel.predict(punto) == model.predict(punto))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Borra el directorio temporal\n",
    "from shutil import rmtree\n",
    "try:\n",
    "     rmtree(path)\n",
    "except OSError:\n",
    "     pass"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.5.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}