Ejercicio_BigDL.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img src=\"./mbit-logo.png\" align=\"right\" style=\"float\" width=\"55\">\n",
    "<font color=\"#CA3532\"><h1 align=\"left\">Ejercicio Spark BigDL</h1></font>\n",
    "<font color=\"#CA3532\"><h2 align=\"left\">Máster en Big Data, Cloud & Analytics 2019-2020</h2></font>\n",
    "<font color=\"#6E6E6E\"><h4 align=\"left\">Carlos Alfonsel <a> carlos.alfonsel@mbitschool.com </a> </h4></font> "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "kCF7FB8dWoaS"
   },
   "source": [
    "# Predicción del precio de casas de Boston\n",
    "\n",
    "Intentaremos predecir el precio mediano de las viviendas en un barrio de Boston a mediados de la década de 1970, partiendo de una variedad de datos sobre el el barrio en el momento, como la tasa de criminalidad, la tasa de impuesto a la propiedad, etc.\n",
    "\n",
    "El conjunto de datos que usaremos es pequeña, 506 muestras en total. Debemos estar atentos, ya que hay valores discretos y continuos (como la criminalidad).\n",
    "\n",
    "**Aunque la solución planteada es con AnalyticsZoo, también se puede utilizar BigDL.**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "zfrSOiiiDFKl"
   },
   "source": [
    "Los datos contienen 13 características a partir de las cuales deberíamos ser capaces de poder predecir el precio de cada vivienda. Las características son:\n",
    "\n",
    "1. Ratio de crimen per cápita.\n",
    "\n",
    "2. Proporción de tierra residencial dividida en zonas de más de 25.000 pies cuadrados.\n",
    "3. Proporción de acres de negocios no minoristas por ciudad.\n",
    "4. Variable ficticia de Charles River (= 1 si limita con el río; 0 en caso contrario).\n",
    "5. Concentración de óxidos nítricos (partes por 10 millones).\n",
    "6. Número promedio de habitaciones por vivienda.\n",
    "7. Proporción de unidades ocupadas por sus propietarios y construídas antes de 1940.\n",
    "8. Distancias ponderadas a cinco centros de empleo de Boston.\n",
    "9. Índice de accesibilidad a carreteras radiales.\n",
    "10. Tasa de impuesto a la propiedad de valor total por $10,000.\n",
    "11. Relación alumno-profesor por localidad.\n",
    "12. 1000 * (Bk - 0.63) ** 2 donde Bk es la proporción de personas negras por pueblo.\n",
    "13. % de población de un estatus bajo."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "KmKVDNcfj9XK",
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "#! pip install analytics-zoo\n",
    "#! sudo apt install openjdk-8-jdk\n",
    "#! echo \"2\" | sudo update-alternatives --config java"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "M86KWgPnlMmF"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Prepending /home/osboxes/anaconda3/envs/env_bigdl/lib/python2.7/site-packages/bigdl/share/conf/spark-bigdl.conf to sys.path\n",
      "Adding /home/osboxes/anaconda3/envs/env_bigdl/lib/python2.7/site-packages/zoo/share/lib/analytics-zoo-bigdl_0.10.0-spark_2.4.3-0.8.1-jar-with-dependencies.jar to BIGDL_JARS\n",
      "Prepending /home/osboxes/anaconda3/envs/env_bigdl/lib/python2.7/site-packages/zoo/share/conf/spark-analytics-zoo.conf to sys.path\n",
      "Adding /home/osboxes/anaconda3/envs/env_bigdl/lib/python2.7/site-packages/zoo/share/lib/analytics-zoo-bigdl_0.10.0-spark_2.4.3-0.8.1-jar-with-dependencies.jar to SPARK_CLASSPATH\n"
     ]
    }
   ],
   "source": [
    "from zoo.common.nncontext import *\n",
    "sc_zoo = init_nncontext(init_spark_conf().setMaster(\"local[4]\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "q1fTd4IuWoaZ"
   },
   "outputs": [],
   "source": [
    "from zoo.pipeline.api.keras.datasets import boston_housing\n",
    "(train_data, train_targets), (test_data, test_targets) =  boston_housing.load_data()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "#! pip install BigDL\n",
    "#! echo \"y\" | pip uninstall pyspark\n",
    "#! pip install pyspark==2.1.2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Populating the interactive namespace from numpy and matplotlib\n"
     ]
    }
   ],
   "source": [
    "import matplotlib\n",
    "matplotlib.use('Agg')\n",
    "%pylab inline\n",
    "import pandas\n",
    "import datetime as dt\n",
    "\n",
    "from bigdl.nn.layer            import *\n",
    "from bigdl.nn.criterion        import *\n",
    "from bigdl.optim.optimizer     import *\n",
    "from bigdl.util.common         import *\n",
    "from bigdl.util.common         import Sample\n",
    "from bigdl.dataset.transformer import *\n",
    "\n",
    "from matplotlib.pyplot         import imshow\n",
    "import matplotlib.pyplot       as plt\n",
    "\n",
    "from pyspark.conf              import SparkConf\n",
    "from pyspark.sql               import SparkSession\n",
    "\n",
    "conf = SparkConf().setMaster(\"local[4]\") \\\n",
    "                  .set(\"spark.driver.memory\",\"2g\") \\\n",
    "                  .set(\"spark.shuffle.reduceLocality.enabled\", \"false\") \\\n",
    "                  .set(\"spark.shuffle.blockTransferService\", \"nio\") \\\n",
    "                  .set(\"spark.scheduler.minRegisteredResourcesRatio\", \"1.0\") \\\n",
    "                  .set(\"spark.speculation\", \"false\")\n",
    "\n",
    "sc = SparkSession.builder.config(conf = conf).getOrCreate().sparkContext\n",
    "\n",
    "init_engine()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(404, 13)\n",
      "(102, 13)\n"
     ]
    }
   ],
   "source": [
    "print(train_data.shape)\n",
    "print(test_data.shape)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "train = Sample.from_ndarray(train_data, train_targets)\n",
    "test  = Sample.from_ndarray(test_data , test_targets)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "#rdd_train = sc.parallelize(range(train_data.shape[0])).map(lambda x: train[x])\n",
    "#for row in rdd_train.collect(): print(row) #NO FUNCIONA CON ESTE TIPO DE VARIABLE (PipelineRDD)\n",
    "\n",
    "rdd_train = sc.parallelize(range(train_data.shape[0])).map(lambda x: \n",
    "                                                           Sample.from_ndarray(train_data   [x], \n",
    "                                                                               train_targets[x]))\n",
    "\n",
    "rdd_test  = sc.parallelize(range(test_data.shape [0])).map(lambda x: \n",
    "                                                          Sample.from_ndarray(test_data    [x], \n",
    "                                                                               test_targets[x]))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "PythonRDD[2] at RDD at PythonRDD.scala:48"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd_train"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[Sample: features: [JTensor: storage: [6.7240e-02 0.0000e+00 3.2400e+00 0.0000e+00 4.6000e-01 6.3330e+00\n",
      " 1.7200e+01 5.2146e+00 4.0000e+00 4.3000e+02 1.6900e+01 3.7521e+02\n",
      " 7.3400e+00], shape: [13], float], labels: [JTensor: storage: [22.6], shape: [1], float], Sample: features: [JTensor: storage: [9.2323e+00 0.0000e+00 1.8100e+01 0.0000e+00 6.3100e-01 6.2160e+00\n",
      " 1.0000e+02 1.1691e+00 2.4000e+01 6.6600e+02 2.0200e+01 3.6615e+02\n",
      " 9.5300e+00], shape: [13], float], labels: [JTensor: storage: [50.], shape: [1], float], Sample: features: [JTensor: storage: [1.1425e-01 0.0000e+00 1.3890e+01 1.0000e+00 5.5000e-01 6.3730e+00\n",
      " 9.2400e+01 3.3633e+00 5.0000e+00 2.7600e+02 1.6400e+01 3.9374e+02\n",
      " 1.0500e+01], shape: [13], float], labels: [JTensor: storage: [23.], shape: [1], float], Sample: features: [JTensor: storage: [ 24.8017   0.      18.1      0.       0.693    5.349   96.       1.7028\n",
      "  24.     666.      20.2    396.9     19.77  ], shape: [13], float], labels: [JTensor: storage: [8.3], shape: [1], float], Sample: features: [JTensor: storage: [5.6460e-02 0.0000e+00 1.2830e+01 0.0000e+00 4.3700e-01 6.2320e+00\n",
      " 5.3700e+01 5.0141e+00 5.0000e+00 3.9800e+02 1.8700e+01 3.8640e+02\n",
      " 1.2340e+01], shape: [13], float], labels: [JTensor: storage: [21.2], shape: [1], float]]\n"
     ]
    }
   ],
   "source": [
    "print rdd_train.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Hyperparameters:\n",
    "learning_rate   = 0.2\n",
    "training_epochs = 15\n",
    "batch_size      = 8\n",
    "n_input         = train_data.shape[1]\n",
    "n_output        = 1\n",
    "\n",
    "# Network Parameters:\n",
    "n_hidden_1 = n_input * 2 # 1st layer number of features\n",
    "n_hidden_2 = n_input * 2 # 2nd layer number of features"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "creating: createSequential\n",
      "creating: createLinear\n",
      "creating: createReLU\n",
      "creating: createLinear\n",
      "creating: createReLU\n",
      "creating: createLinear\n",
      "creating: createLogSoftMax\n"
     ]
    }
   ],
   "source": [
    "# Linear Regression Model:\n",
    "def linear_regression(n_input, n_output):\n",
    "    model = Sequential()  \n",
    "    model.add(Linear(n_input, n_output))\n",
    "    return model\n",
    "\n",
    "#model = linear_regression(n_input, n_output)\n",
    "\n",
    "# MLP Model:\n",
    "def multilayer_perceptron(n_hidden_1, n_hidden_2, n_input, n_output):\n",
    "    # Initialize a sequential container:\n",
    "    model = Sequential()\n",
    "    # Hidden layer with ReLu activation:\n",
    "    model.add(Linear(n_input   , n_hidden_1).set_name('mlp_fc1'))\n",
    "    model.add(ReLU())\n",
    "    # Hidden layer with ReLu activation:\n",
    "    model.add(Linear(n_hidden_1, n_hidden_2).set_name('mlp_fc2'))\n",
    "    model.add(ReLU())\n",
    "    # Output layer:\n",
    "    model.add(Linear(n_hidden_2, n_output  ).set_name('mlp_fc3'))\n",
    "    model.add(LogSoftMax())\n",
    "    return model\n",
    "\n",
    "model = multilayer_perceptron(n_hidden_1, n_hidden_2, n_input, n_output)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "creating: createMSECriterion\n",
      "creating: createAbsCriterion\n",
      "creating: createSmoothL1Criterion\n",
      "creating: createBCECriterion\n",
      "creating: createDefault\n",
      "creating: createSGD\n",
      "creating: createMaxEpoch\n",
      "creating: createDistriOptimizer\n"
     ]
    }
   ],
   "source": [
    "# Optimizer:\n",
    "\n",
    "criterion_1 = MSECriterion()\n",
    "criterion_2 = AbsCriterion()\n",
    "criterion_3 = SmoothL1Criterion()\n",
    "criterion_4 = BCECriterion()\n",
    "\n",
    "optimizer   = Optimizer(model        = model,\n",
    "                        training_rdd = rdd_train,\n",
    "                        criterion    = criterion_4,\n",
    "                        optim_method = SGD(learningrate = learning_rate),\n",
    "                        end_trigger  = MaxEpoch(training_epochs),\n",
    "                        batch_size   = batch_size)\n",
    "\n",
    "# bigdl.nn.criterion:\n",
    "# https://bigdl-project.github.io/0.2.0/APIdocs/python-api-doc/_modules/bigdl/nn/criterion.html\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Training model:\n",
    "trained_model = optimizer.optimize()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "predict: \n",
      "\n",
      "[0.]\n",
      "\n",
      "[0.]\n",
      "\n",
      "[0.]\n",
      "\n",
      "[0.]\n",
      "\n",
      "[0.]\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Printing the first 10 predicted results of training data:\n",
    "predict_result = trained_model.predict(rdd_train)\n",
    "\n",
    "p = predict_result.take(5)\n",
    "print(\"predict: \\n\")\n",
    "for i in p: print(str(i) + \"\\n\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[Sample: features: [JTensor: storage: [3.7380e-02 0.0000e+00 5.1900e+00 0.0000e+00 5.1500e-01 6.3100e+00\n",
      " 3.8500e+01 6.4584e+00 5.0000e+00 2.2400e+02 2.0200e+01 3.8940e+02\n",
      " 6.7500e+00], shape: [13], float], labels: [JTensor: storage: [20.7], shape: [1], float], Sample: features: [JTensor: storage: [6.5880e-02 0.0000e+00 2.4600e+00 0.0000e+00 4.8800e-01 7.7650e+00\n",
      " 8.3300e+01 2.7410e+00 3.0000e+00 1.9300e+02 1.7800e+01 3.9556e+02\n",
      " 7.5600e+00], shape: [13], float], labels: [JTensor: storage: [39.8], shape: [1], float], Sample: features: [JTensor: storage: [  8.98296   0.       18.1       1.        0.77      6.212    97.4\n",
      "   2.1222   24.      666.       20.2     377.73     17.6    ], shape: [13], float], labels: [JTensor: storage: [17.8], shape: [1], float], Sample: features: [JTensor: storage: [  1.19294   0.       21.89      0.        0.624     6.326    97.7\n",
      "   2.271     4.      437.       21.2     396.9      12.26   ], shape: [13], float], labels: [JTensor: storage: [19.6], shape: [1], float], Sample: features: [JTensor: storage: [  9.51363   0.       18.1       0.        0.713     6.728    94.1\n",
      "   2.4961   24.      666.       20.2       6.68     18.71   ], shape: [13], float], labels: [JTensor: storage: [14.9], shape: [1], float]]\n"
     ]
    }
   ],
   "source": [
    "predict_data = sc.parallelize(range(test_data.shape[0])).map(lambda x: \n",
    "                                                             Sample.from_ndarray(test_data   [x],\n",
    "                                                                                 test_targets[x]))\n",
    "print predict_data.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Testing the model based on MSE (Mean Squared Error):\n",
    "def test_predict(trained_model):\n",
    "    total_length = test_data.shape[0]\n",
    "    features     = test_data\n",
    "    labels       = test_targets\n",
    "    predict_data = sc.parallelize(range(total_length)).map(lambda x: \n",
    "                                                           Sample.from_ndarray(features[x], \n",
    "                                                                               labels  [x]))\n",
    "    \n",
    "    predict_result = trained_model.predict(predict_data)\n",
    "    \n",
    "    p   = sum(predict_result.collect())\n",
    "    mse = ((p - sum(labels)) ** 2) / total_length\n",
    "    mae = abs((p - sum(labels)) / total_length)\n",
    "    print(\"Mean Squared Error: \", mse)\n",
    "    print(\"Mean Absolute Error:\", mae)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "('Mean Squared Error: ', 52736.789117647044)\n",
      "('Mean Absolute Error:', 22.738235294117644)\n"
     ]
    }
   ],
   "source": [
    "test_predict(trained_model)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "colab": {
   "collapsed_sections": [],
   "name": "Predecir el precio de la vivienda - Sin solución.ipynb",
   "provenance": []
  },
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.18"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}