Analisi dei dati in PySpark.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Analisi dei dati in PySpark\n",
    "\n",
    "Utilizzeremo PySpark attraverso un Notebook, usando il pacchetto `findspark` installato tramite `conda`. Anche PySpark dovrà essere installato, attraverso conda ovvero scaricandolo dalla sua [home page](https://spark.apache.org/downloads.html).\n",
    "\n",
    "Sarà necessario definire la variabile di ambiente `$SPARK_HOME` che punta al percorso di installazione di PySpark.\n",
    "\n",
    "Il pacchetto `findspark` ci fornisce un semplice modo per inizializzare una `SparkSession` all'interno di un qualunque IDE, invocando poi il package `pyspark` e tutti i suoi moduli.\n",
    "\n",
    "Assumendo che l'installazione di Python e Apache Hadoop sia già stata fatta correttamente e che sia stata definita `$SPARK_HOME`, basta usare il seguente prologo all'inizio di ogni applicazione Pyspark:\n",
    "\n",
    "```python\n",
    "import findspark\n",
    "location = findspark.find()\n",
    "findspark.init(location)\n",
    "\n",
    "import pyspark\n",
    "from pyspark.sql import SparkSession\n",
    "\n",
    "spark = SparkSession.builder.appName('MyAppName').getOrCreate()\n",
    "\n",
    "```\n",
    "\n",
    "La ``SparkSession`` ci fornisce l'accesso alla _DataFrame API_ oltre che la possibilità di accedere ai diversi componenti del framework:\n",
    "\n",
    "- `SparkContext`\n",
    "- `SQLContext`\n",
    "- `StreamingContext`\n",
    "- `HiveContext`\n",
    "\n",
    "In particolare, lo ``SparkContext`` consente l'accesso alle _RDD API_ e si potrà ottenere con il seguente codice\n",
    "\n",
    "```python\n",
    "sc = spark.sparkContext\n",
    "\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Creiamo la SparkSession per la nostra applicazione\n",
    "\n",
    "import findspark\n",
    "\n",
    "location = findspark.find()\n",
    "findspark.init(location)\n",
    "\n",
    "import pyspark\n",
    "\n",
    "from pyspark.sql import SparkSession\n",
    "spark = SparkSession.builder.appName('SurvivedPassengers').getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "\n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"http://192.168.223.100:4040\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v3.0.1</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local[*]</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>SurvivedPassengers</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        "
      ],
      "text/plain": [
       "<SparkContext master=local[*] appName=SurvivedPassengers>"
      ]
     },
     "execution_count": 30,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Lo SparkContext sarà descritto come segue\n",
    "\n",
    "spark.sparkContext"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Importazione dei dati dal data set titanic.csv\n",
    "from pyspark.sql.types import *\n",
    "\n",
    "# costruzione dello schema del DataFrame\n",
    "\n",
    "passengerSchema = StructType([\n",
    "    StructField('PassengerID',ShortType(),False),\n",
    "    StructField('Survived',ShortType(),False),\n",
    "    StructField('Pclass',ShortType(),False),\n",
    "    StructField('Name',StringType(),False),\n",
    "    StructField('Sex',StringType(),False),\n",
    "    StructField('Age',FloatType(),True),\n",
    "    StructField('SibSp',IntegerType(),True),\n",
    "    StructField('Parch',IntegerType(),True),\n",
    "    StructField('Ticket',StringType(),True),\n",
    "    StructField('Fare',FloatType(),True),\n",
    "    StructField('Cabin',StringType(),True),\n",
    "    StructField('Embarked',StringType(),True)\n",
    "])\n",
    "\n",
    "titanic = spark.read.format('csv')\\\n",
    "    .option('header','true')\\\n",
    "    .option('mode','FAILFAST')\\\n",
    "    .schema(passengerSchema)\\\n",
    "    .load('hdfs://localhost:9099/user/pirrone/spark/input/titanic.csv')\n",
    "                    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- PassengerID: short (nullable = true)\n",
      " |-- Survived: short (nullable = true)\n",
      " |-- Pclass: short (nullable = true)\n",
      " |-- Name: string (nullable = true)\n",
      " |-- Sex: string (nullable = true)\n",
      " |-- Age: float (nullable = true)\n",
      " |-- SibSp: integer (nullable = true)\n",
      " |-- Parch: integer (nullable = true)\n",
      " |-- Ticket: string (nullable = true)\n",
      " |-- Fare: float (nullable = true)\n",
      " |-- Cabin: string (nullable = true)\n",
      " |-- Embarked: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "titanic.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+--------+------+--------------------+------+----+-----+-----+----------------+-------+-----+--------+\n",
      "|PassengerID|Survived|Pclass|                Name|   Sex| Age|SibSp|Parch|          Ticket|   Fare|Cabin|Embarked|\n",
      "+-----------+--------+------+--------------------+------+----+-----+-----+----------------+-------+-----+--------+\n",
      "|          1|       0|     3|Braund, Mr. Owen ...|  male|22.0|    1|    0|       A/5 21171|   7.25| null|       S|\n",
      "|          2|       1|     1|Cumings, Mrs. Joh...|female|38.0|    1|    0|        PC 17599|71.2833|  C85|       C|\n",
      "|          3|       1|     3|Heikkinen, Miss. ...|female|26.0|    0|    0|STON/O2. 3101282|  7.925| null|       S|\n",
      "|          4|       1|     1|Futrelle, Mrs. Ja...|female|35.0|    1|    0|          113803|   53.1| C123|       S|\n",
      "|          5|       0|     3|Allen, Mr. Willia...|  male|35.0|    0|    0|          373450|   8.05| null|       S|\n",
      "|          6|       0|     3|    Moran, Mr. James|  male|null|    0|    0|          330877| 8.4583| null|       Q|\n",
      "|          7|       0|     1|McCarthy, Mr. Tim...|  male|54.0|    0|    0|           17463|51.8625|  E46|       S|\n",
      "|          8|       0|     3|Palsson, Master. ...|  male| 2.0|    3|    1|          349909| 21.075| null|       S|\n",
      "|          9|       1|     3|Johnson, Mrs. Osc...|female|27.0|    0|    2|          347742|11.1333| null|       S|\n",
      "|         10|       1|     2|Nasser, Mrs. Nich...|female|14.0|    1|    0|          237736|30.0708| null|       C|\n",
      "|         11|       1|     3|Sandstrom, Miss. ...|female| 4.0|    1|    1|         PP 9549|   16.7|   G6|       S|\n",
      "|         12|       1|     1|Bonnell, Miss. El...|female|58.0|    0|    0|          113783|  26.55| C103|       S|\n",
      "|         13|       0|     3|Saundercock, Mr. ...|  male|20.0|    0|    0|       A/5. 2151|   8.05| null|       S|\n",
      "|         14|       0|     3|Andersson, Mr. An...|  male|39.0|    1|    5|          347082| 31.275| null|       S|\n",
      "|         15|       0|     3|Vestrom, Miss. Hu...|female|14.0|    0|    0|          350406| 7.8542| null|       S|\n",
      "|         16|       1|     2|Hewlett, Mrs. (Ma...|female|55.0|    0|    0|          248706|   16.0| null|       S|\n",
      "|         17|       0|     3|Rice, Master. Eugene|  male| 2.0|    4|    1|          382652| 29.125| null|       Q|\n",
      "|         18|       1|     2|Williams, Mr. Cha...|  male|null|    0|    0|          244373|   13.0| null|       S|\n",
      "|         19|       0|     3|Vander Planke, Mr...|female|31.0|    1|    0|          345763|   18.0| null|       S|\n",
      "|         20|       1|     3|Masselmani, Mrs. ...|female|null|    0|    0|            2649|  7.225| null|       C|\n",
      "+-----------+--------+------+--------------------+------+----+-----+-----+----------------+-------+-----+--------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "titanic.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [],
   "source": [
    "# imbarcati\n",
    "\n",
    "from pyspark.sql.functions import *\n",
    "\n",
    "passengers = titanic.groupBy(expr('Sex as Gender'),expr('Pclass as Class'))\\\n",
    "                    .agg(expr('count(1) as pass_num'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-----+--------+\n",
      "|Gender|Class|pass_num|\n",
      "+------+-----+--------+\n",
      "|  male|    3|     347|\n",
      "|female|    3|     144|\n",
      "|female|    1|      94|\n",
      "|female|    2|      76|\n",
      "|  male|    2|     108|\n",
      "|  male|    1|     122|\n",
      "+------+-----+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "passengers.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [],
   "source": [
    "#sopravvissuti\n",
    "\n",
    "survived = titanic.filter('Survived == 1').groupBy('Sex','Pclass')\\\n",
    "            .agg(expr('avg(Age)'),expr('count(1) as surv_num'))\\\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+------+------------------+--------+\n",
      "|   Sex|Pclass|          avg(Age)|surv_num|\n",
      "+------+------+------------------+--------+\n",
      "|  male|     3| 22.27421052597071|      47|\n",
      "|female|     3|19.329787234042552|      72|\n",
      "|female|     1|  34.9390243902439|      91|\n",
      "|female|     2|28.080882352941178|      70|\n",
      "|  male|     2| 16.02199999888738|      17|\n",
      "|  male|     1|36.248000000417235|      45|\n",
      "+------+------+------------------+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "survived.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Inner join tra passeggeri e sopravvisuti sulla chiave di raggruppamento\n",
    "# drop delle colonne di chiave doppie\n",
    "# creazione della colonna della percentuale di sopravvivenza\n",
    "\n",
    "survivedPassengers = passengers.join(survived,\n",
    "             (survived.Sex == passengers.Gender) & (survived.Pclass == passengers.Class),\n",
    "              'inner')\\\n",
    "        .drop('Sex','Pclass')\\\n",
    "        .withColumn('surv_perc',100.0 * col('surv_num') / col('pass_num'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-----+--------+------------------+--------+------------------+\n",
      "|Gender|Class|pass_num|          avg(Age)|surv_num|         surv_perc|\n",
      "+------+-----+--------+------------------+--------+------------------+\n",
      "|  male|    3|     347| 22.27421052597071|      47|13.544668587896254|\n",
      "|female|    3|     144|19.329787234042552|      72|              50.0|\n",
      "|female|    1|      94|  34.9390243902439|      91| 96.80851063829788|\n",
      "|female|    2|      76|28.080882352941178|      70| 92.10526315789474|\n",
      "|  male|    2|     108| 16.02199999888738|      17| 15.74074074074074|\n",
      "|  male|    1|     122|36.248000000417235|      45|36.885245901639344|\n",
      "+------+-----+--------+------------------+--------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "survivedPassengers.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Salvataggio del file survived.tsv nel file system\n",
    "\n",
    "survivedPassengers.write\\\n",
    "                  .format('csv')\\\n",
    "                  .mode('overwrite')\\\n",
    "                  .option('sep','\\t')\\\n",
    "                  .save('hdfs://localhost:9099/user/pirrone/spark/output/survived.tsv')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Utilizzo delle API SQL\n",
    "# Creazione delle tabelle temporanee\n",
    "\n",
    "titanic.createOrReplaceTempView('titanicTable')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {},
   "outputs": [],
   "source": [
    "passSQL = spark.sql(\"\"\"\n",
    "SELECT Sex AS Gender, Pclass AS Class, Count(1) as Pass_num FROM titanicTable \n",
    "GROUP BY Sex, PClass\n",
    "\"\"\")\n",
    "passSQL.createOrReplaceTempView('passTable')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-----+--------+\n",
      "|Gender|Class|Pass_num|\n",
      "+------+-----+--------+\n",
      "|  male|    3|     347|\n",
      "|female|    3|     144|\n",
      "|female|    1|      94|\n",
      "|female|    2|      76|\n",
      "|  male|    2|     108|\n",
      "|  male|    1|     122|\n",
      "+------+-----+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "passSQL.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {},
   "outputs": [],
   "source": [
    "survSQL = spark.sql(\"\"\"\n",
    "SELECT Sex, Pclass, AVG(Age) as Avg_age, Count(1) as Surv_num FROM titanicTable\n",
    "WHERE Survived = 1\n",
    "GROUP BY Sex, PClass\n",
    "\"\"\")\n",
    "survSQL.createOrReplaceTempView('survTable')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+------+------------------+--------+\n",
      "|   Sex|Pclass|           Avg_age|Surv_num|\n",
      "+------+------+------------------+--------+\n",
      "|  male|     3| 22.27421052597071|      47|\n",
      "|female|     3|19.329787234042552|      72|\n",
      "|female|     1|  34.9390243902439|      91|\n",
      "|female|     2|28.080882352941178|      70|\n",
      "|  male|     2| 16.02199999888738|      17|\n",
      "|  male|     1|36.248000000417235|      45|\n",
      "+------+------+------------------+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "survSQL.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {},
   "outputs": [],
   "source": [
    "survPercSQL = spark.sql(\"\"\"\n",
    "SELECT passTable.Gender, passTable.Class, passTable.Pass_num, \n",
    "survTable.Avg_age, survTable.Surv_num, \n",
    "100.0 * survTable.Surv_num / passTable.Pass_num AS Surv_perc\n",
    "FROM passTable\n",
    "INNER JOIN survTable \n",
    "ON (passTable.Gender = survTable.Sex AND passTable.Class = survTable.PClass)\n",
    "\"\"\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-----+--------+------------------+--------+-----------------+\n",
      "|Gender|Class|Pass_num|           Avg_age|Surv_num|        Surv_perc|\n",
      "+------+-----+--------+------------------+--------+-----------------+\n",
      "|  male|    3|     347| 22.27421052597071|      47|13.54466858789625|\n",
      "|female|    3|     144|19.329787234042552|      72|50.00000000000000|\n",
      "|female|    1|      94|  34.9390243902439|      91|96.80851063829787|\n",
      "|female|    2|      76|28.080882352941178|      70|92.10526315789474|\n",
      "|  male|    2|     108| 16.02199999888738|      17|15.74074074074074|\n",
      "|  male|    1|     122|36.248000000417235|      45|36.88524590163934|\n",
      "+------+-----+--------+------------------+--------+-----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "survPercSQL.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Uso della RDD API\n",
    "# Estraiamo il RDD dal DataFrame titanic\n",
    "\n",
    "titanicRDD = titanic.rdd"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(PassengerID=1, Survived=0, Pclass=3, Name='Braund, Mr. Owen Harris', Sex='male', Age=22.0, SibSp=1, Parch=0, Ticket='A/5 21171', Fare=7.25, Cabin=None, Embarked='S'),\n",
       " Row(PassengerID=2, Survived=1, Pclass=1, Name='Cumings, Mrs. John Bradley (Florence Briggs Thayer)', Sex='female', Age=38.0, SibSp=1, Parch=0, Ticket='PC 17599', Fare=71.2833023071289, Cabin='C85', Embarked='C'),\n",
       " Row(PassengerID=3, Survived=1, Pclass=3, Name='Heikkinen, Miss. Laina', Sex='female', Age=26.0, SibSp=0, Parch=0, Ticket='STON/O2. 3101282', Fare=7.925000190734863, Cabin=None, Embarked='S'),\n",
       " Row(PassengerID=4, Survived=1, Pclass=1, Name='Futrelle, Mrs. Jacques Heath (Lily May Peel)', Sex='female', Age=35.0, SibSp=1, Parch=0, Ticket='113803', Fare=53.099998474121094, Cabin='C123', Embarked='S'),\n",
       " Row(PassengerID=5, Survived=0, Pclass=3, Name='Allen, Mr. William Henry', Sex='male', Age=35.0, SibSp=0, Parch=0, Ticket='373450', Fare=8.050000190734863, Cabin=None, Embarked='S'),\n",
       " Row(PassengerID=6, Survived=0, Pclass=3, Name='Moran, Mr. James', Sex='male', Age=None, SibSp=0, Parch=0, Ticket='330877', Fare=8.45829963684082, Cabin=None, Embarked='Q'),\n",
       " Row(PassengerID=7, Survived=0, Pclass=1, Name='McCarthy, Mr. Timothy J', Sex='male', Age=54.0, SibSp=0, Parch=0, Ticket='17463', Fare=51.86249923706055, Cabin='E46', Embarked='S'),\n",
       " Row(PassengerID=8, Survived=0, Pclass=3, Name='Palsson, Master. Gosta Leonard', Sex='male', Age=2.0, SibSp=3, Parch=1, Ticket='349909', Fare=21.075000762939453, Cabin=None, Embarked='S'),\n",
       " Row(PassengerID=9, Survived=1, Pclass=3, Name='Johnson, Mrs. Oscar W (Elisabeth Vilhelmina Berg)', Sex='female', Age=27.0, SibSp=0, Parch=2, Ticket='347742', Fare=11.133299827575684, Cabin=None, Embarked='S'),\n",
       " Row(PassengerID=10, Survived=1, Pclass=2, Name='Nasser, Mrs. Nicholas (Adele Achem)', Sex='female', Age=14.0, SibSp=1, Parch=0, Ticket='237736', Fare=30.07080078125, Cabin=None, Embarked='C'),\n",
       " Row(PassengerID=11, Survived=1, Pclass=3, Name='Sandstrom, Miss. Marguerite Rut', Sex='female', Age=4.0, SibSp=1, Parch=1, Ticket='PP 9549', Fare=16.700000762939453, Cabin='G6', Embarked='S'),\n",
       " Row(PassengerID=12, Survived=1, Pclass=1, Name='Bonnell, Miss. Elizabeth', Sex='female', Age=58.0, SibSp=0, Parch=0, Ticket='113783', Fare=26.549999237060547, Cabin='C103', Embarked='S'),\n",
       " Row(PassengerID=13, Survived=0, Pclass=3, Name='Saundercock, Mr. William Henry', Sex='male', Age=20.0, SibSp=0, Parch=0, Ticket='A/5. 2151', Fare=8.050000190734863, Cabin=None, Embarked='S'),\n",
       " Row(PassengerID=14, Survived=0, Pclass=3, Name='Andersson, Mr. Anders Johan', Sex='male', Age=39.0, SibSp=1, Parch=5, Ticket='347082', Fare=31.274999618530273, Cabin=None, Embarked='S'),\n",
       " Row(PassengerID=15, Survived=0, Pclass=3, Name='Vestrom, Miss. Hulda Amanda Adolfina', Sex='female', Age=14.0, SibSp=0, Parch=0, Ticket='350406', Fare=7.8541998863220215, Cabin=None, Embarked='S'),\n",
       " Row(PassengerID=16, Survived=1, Pclass=2, Name='Hewlett, Mrs. (Mary D Kingcome) ', Sex='female', Age=55.0, SibSp=0, Parch=0, Ticket='248706', Fare=16.0, Cabin=None, Embarked='S'),\n",
       " Row(PassengerID=17, Survived=0, Pclass=3, Name='Rice, Master. Eugene', Sex='male', Age=2.0, SibSp=4, Parch=1, Ticket='382652', Fare=29.125, Cabin=None, Embarked='Q'),\n",
       " Row(PassengerID=18, Survived=1, Pclass=2, Name='Williams, Mr. Charles Eugene', Sex='male', Age=None, SibSp=0, Parch=0, Ticket='244373', Fare=13.0, Cabin=None, Embarked='S'),\n",
       " Row(PassengerID=19, Survived=0, Pclass=3, Name='Vander Planke, Mrs. Julius (Emelia Maria Vandemoortele)', Sex='female', Age=31.0, SibSp=1, Parch=0, Ticket='345763', Fare=18.0, Cabin=None, Embarked='S'),\n",
       " Row(PassengerID=20, Survived=1, Pclass=3, Name='Masselmani, Mrs. Fatima', Sex='female', Age=None, SibSp=0, Parch=0, Ticket='2649', Fare=7.224999904632568, Cabin=None, Embarked='C')]"
      ]
     },
     "execution_count": 48,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# titanicRDD è una lista di named tuple di classe Row\n",
    "# mostriamo i primi 20 oggetti\n",
    "\n",
    "titanicRDD.take(20)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {},
   "outputs": [],
   "source": [
    "# mappiamo il RDD titanic per estrarre solo le colonne Sex, Pclass, Age\n",
    "# e creiamo la chiave (Sex, Pclass)\n",
    "\n",
    "keyPass = titanicRDD\\\n",
    "            .keyBy(lambda row: (row.Sex, row.Pclass))\\\n",
    "            .map(lambda row: (row[0], row[1].Age))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(('male', 3), 22.0),\n",
       " (('female', 1), 38.0),\n",
       " (('female', 3), 26.0),\n",
       " (('female', 1), 35.0),\n",
       " (('male', 3), 35.0)]"
      ]
     },
     "execution_count": 50,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "keyPass.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "defaultdict(int,\n",
       "            {('male', 3): 347,\n",
       "             ('female', 1): 94,\n",
       "             ('female', 3): 144,\n",
       "             ('male', 1): 122,\n",
       "             ('female', 2): 76,\n",
       "             ('male', 2): 108})"
      ]
     },
     "execution_count": 51,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Contiamo per chiave e otteniamo i passeggeri imbarcati per classe e genere\n",
    "\n",
    "embarkedPass = keyPass.countByKey()\n",
    "\n",
    "embarkedPass"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Analogo processing per i sopravvissuti, ma\n",
    "# prima filtriamo per Survived == 1\n",
    "\n",
    "keySurv = titanicRDD\\\n",
    "            .filter(lambda row: row.Survived == 1)\\\n",
    "            .keyBy(lambda row: (row.Sex, row.Pclass))\\\n",
    "            .map(lambda row: (row[0], row[1].Age))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(('female', 1), 38.0),\n",
       " (('female', 3), 26.0),\n",
       " (('female', 1), 35.0),\n",
       " (('female', 3), 27.0),\n",
       " (('female', 2), 14.0),\n",
       " (('female', 3), 4.0),\n",
       " (('female', 1), 58.0),\n",
       " (('female', 2), 55.0),\n",
       " (('male', 2), None),\n",
       " (('female', 3), None)]"
      ]
     },
     "execution_count": 53,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "keySurv.take(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "defaultdict(int,\n",
       "            {('female', 1): 91,\n",
       "             ('female', 3): 72,\n",
       "             ('female', 2): 70,\n",
       "             ('male', 2): 17,\n",
       "             ('male', 1): 45,\n",
       "             ('male', 3): 47})"
      ]
     },
     "execution_count": 54,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "survPass = keySurv.countByKey()\n",
    "\n",
    "survPass"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(('female', 1), 91, 96.80851063829788, 31.483516483516482),\n",
       " (('female', 3), 72, 50.0, 12.618055555555555),\n",
       " (('female', 2), 70, 92.10526315789474, 27.27857142857143),\n",
       " (('male', 2), 17, 15.74074074074074, 14.137058822547688),\n",
       " (('male', 1), 45, 36.885245901639344, 32.22044444481532),\n",
       " (('male', 3), 47, 13.544668587896254, 18.008936169933765)]"
      ]
     },
     "execution_count": 55,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# effettuiamo il reduce per sommare le età\n",
    "# con una funzione di somma che esclude i valori nulli\n",
    "# poi effettuiamo il seguente map\n",
    "# (<chiave>, <somma_eta>) --> \n",
    "# (<chiave>, <num_sopravvissuti>, \n",
    "#            <num_sopravvissuti>/<num_imbarcati>, \n",
    "#            <somma_eta>/<num_imbarcati>)\n",
    "\n",
    "def addNoNull(x,y):\n",
    "    if(x == None):\n",
    "        return y\n",
    "    elif(y == None):\n",
    "        return x\n",
    "    else:\n",
    "        return x+y\n",
    "\n",
    "keySurv.reduceByKey(addNoNull)\\\n",
    "        .map(lambda row: (row[0], \n",
    "                          survPass[row[0]], \n",
    "                          100.0*survPass[row[0]]/embarkedPass[row[0]], \n",
    "                          row[1]/survPass[row[0]]))\\\n",
    "        .collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3.7.8 ('bigdata_py37')",
   "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.7.8"
  },
  "vscode": {
   "interpreter": {
    "hash": "f7fdaf848b1569d220adb27db25f0581f160458c042f09ff1ba22ef71dc5af32"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}