Tutorial Apache Spark: ML con PySpark

Scaricare pyspark con pip

Poi puoi scaricare e installare PySpark con l’aiuto di pip. È piuttosto semplice, come per qualsiasi altro pacchetto. Esegui il solito comando e il grosso del lavoro verrà fatto per te:

$ pip install pyspark

In alternativa, puoi anche andare alla pagina di download di Spark. Mantieni le opzioni predefinite nei primi tre passaggi e troverai un link scaricabile al passaggio 4. Clicca su quel link per scaricare. Per questo tutorial, scaricherai la Release 2.2.0 di Spark e il pacchetto “Pre-built for Apache Hadoop 2.7 and later”.

Nota che il download può richiedere un po’ di tempo!

Scaricare Spark con Homebrew

Puoi anche installare Spark con Homebrew, un package manager gratuito e open source. È particolarmente comodo se lavori con macOS.

Esegui semplicemente i seguenti comandi per cercare Spark, ottenere maggiori informazioni e infine installarlo sul tuo computer:

# Search for spark
$ brew search spark

# Get more information on apache-spark
$ brew info apache-spark

# Install apache-spark
$ brew install apache-spark

Download e configurazione di Spark

Poi, assicurati di scompattare la directory che compare nella tua cartella Downloads. Questo può avvenire automaticamente facendo doppio clic sull’archivio spark-2.2.0-bin-hadoop2.7.tgz oppure aprendo il Terminale ed eseguendo il seguente comando:

$ tar xvf spark-2.2.0-bin-hadoop2.7.tgz

Successivamente, sposta la cartella estratta in /usr/local/spark eseguendo la seguente riga:

$ mv spark-2.1.0-bin-hadoop2.7 /usr/local/spark

Nota che se ricevi un errore di permessi negati nello spostare la cartella nella nuova posizione, dovresti aggiungere sudo davanti a questo comando. La riga sopra diventerà quindi $ sudo mv spark-2.1.0-bin-hadoop2.7 /usr/local/spark. Ti verrà chiesta la password, che di solito è la stessa che usi per sbloccare il PC all’avvio :)

Ora che è tutto pronto, apri il file README nel percorso /usr/local/spark. Puoi farlo eseguendo

$ cd /usr/local/spark

Questo ti porta nella cartella che ti serve. Poi puoi iniziare a ispezionare la cartella e a leggere il file README incluso.

Per prima cosa, usa $ ls per ottenere l’elenco di file e cartelle presenti in questa cartella spark. Vedrai che c’è un file README.md. Puoi aprirlo eseguendo uno dei seguenti comandi:

# Open and edit the file
$ nano README.md

# Just read the file 
$ cat README.md

Consiglio usa il tasto tab sulla tastiera per l’autocompletamento mentre digiti il nome del file :) Ti farà risparmiare tempo.

Vedrai che questo README fornisce informazioni generali su Spark, documentazione online, build di Spark, le shell interattive Scala e Python, programmi di esempio e molto altro.

La parte che potrebbe interessarti di più è la sezione su come effettuare la build di Spark, ma nota che questo è particolarmente rilevante solo se non hai scaricato una versione precompilata. Per questo tutorial, invece, hai scaricato una versione precompilata. Puoi premere CTRL + X per uscire dal README, tornando alla cartella spark.

Nel caso avessi selezionato una versione non ancora compilata, assicurati di eseguire il comando elencato nel file README. Al momento della stesura, è il seguente:

$ build/mvn -DskipTests clean package run

Nota che questo comando può impiegare un po’ a terminare.

Applicazioni Spark vs Spark Shell

La shell interattiva è un esempio di ambiente Read-Eval(uate)-Print-Loop (REPL); Significa che tutto ciò che digiti viene letto, valutato e stampato così puoi proseguire l’analisi. Questo potrebbe ricordarti IPython, una potente shell interattiva Python che potresti conoscere da Jupyter. Se vuoi saperne di più, leggi il post di DataCamp IPython o Jupyter.

Ciò significa che puoi usare la shell, disponibile sia per Python che per Scala, per tutto il lavoro interattivo di cui hai bisogno.

Oltre alla shell, puoi anche scrivere e distribuire applicazioni Spark. A differenza della scrittura di applicazioni Spark, nella shell SparkSession è già stata creata per te così puoi iniziare subito a lavorare e non perdere tempo prezioso a crearla.

Ora potresti chiederti: cos’è la SparkSession?

È il punto di ingresso principale alle funzionalità di Spark: rappresenta la connessione a un cluster Spark e puoi usarla per creare RDD e per fare broadcast di variabili su quel cluster. Quando lavori con Spark, tutto inizia e finisce con questa SparkSession. Nota che prima di Spark 2.0.0 i tre principali oggetti di connessione erano SparkContext, SqlContext e HiveContext.

Vedrai altro su questo più avanti. Per ora concentriamoci sulla shell.

La Spark Shell in Python

Dalla cartella spark in /usr/local/spark, puoi eseguire

$ ./bin/pyspark

All’inizio vedrai comparire del testo. Poi apparirà “Spark”, proprio così:

Python 2.7.13 (v2.7.13:a06454b1afa1, Dec 17 2016, 12:39:47) 
[GCC 4.2.1 (Apple Inc. build 5666) (dot 3)] on darwin
Type "help", "copyright", "credits" or "license" for more information.
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
17/07/26 11:41:26 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
17/07/26 11:41:47 WARN ObjectStore: Failed to get database global_temp, returning NoSuchObjectException
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /__ / .__/\_,_/_/ /_/\_\   version 2.2.0
      /_/

Using Python version 2.7.13 (v2.7.13:a06454b1afa1, Dec 17 2016 12:39:47)
SparkSession available as 'spark'.
>>>

Quando vedi questo, sai che sei pronto per iniziare a sperimentare nella shell interattiva!

Consiglio: se preferisci usare la shell IPython invece della Spark shell, puoi farlo impostando la seguente variabile d’ambiente:

export PYSPARK_DRIVER_PYTHON="/usr/local/ipython/bin/ipython"

>>> rdd1 = spark.sparkContext.parallelize([('a',7),('a',2),('b',2)])
>>> rdd2 = spark.sparkContext.parallelize([("a",["x","y","z"]), ("b",["p", "r"])])
>>> rdd3 = spark.sparkContext.parallelize(range(100))

>>> rdd1.reduce(lambda a,b: a+b)

>>> rdd2.flatMapValues(lambda x: x).collect()
[('a', 'x'), ('a', 'y'), ('a', 'z'), ('b', 'p'), ('b', 'r')]

PySpark in Jupyter Notebook

Per questa parte del tutorial non userai la ishell ma costruirai la tua applicazione. Lo farai in un Jupyter Notebook. Hai già tutto il necessario installato, quindi non devi fare molto per far funzionare PySpark in Jupyter.

Puoi lanciare l’applicazione notebook come sempre, eseguendo $ jupyter notebook. Poi crei un nuovo notebook e semplicemente importi la libreria findspark e usi la funzione init(). In questo caso, fornirai il percorso /usr/local/spark a init() perché sei certo che sia il percorso in cui hai installato Spark.

# Import findspark 
import findspark

# Initialize and provide path
findspark.init("/usr/local/spark")

# Or use this alternative
#findspark.init()

Consiglio: se non hai idea se il tuo percorso sia impostato correttamente o dove hai installato Spark sul PC, puoi sempre usare findspark.find() per rilevare automaticamente la posizione dell’installazione di Spark.

Se cerchi modi alternativi per lavorare con Spark in Jupyter, consulta la nostra Guida introduttiva ad Apache Spark in Python.

Ora che hai sistemato tutto, puoi finalmente iniziare creando il tuo primo programma Spark!

Creare il tuo primo programma Spark

Per prima cosa vuoi importare SparkContext dal pacchetto pyspark e inizializzarlo. Ricorda che prima non dovevi farlo perché la shell interattiva di Spark lo creava e inizializzava automaticamente per te! Qui dovrai fare un po’ più di lavoro da solo :)

Importa il modulo SparkSession da pyspark.sql e costruisci una SparkSession con il metodo builder(). Successivamente puoi impostare l’URL del master a cui connetterti, il nome dell’applicazione, aggiungere qualche configurazione aggiuntiva come la memoria dell’executor e infine usare getOrCreate() per ottenere la sessione Spark corrente o crearne una se non ce n’è nessuna in esecuzione.

# Import SparkSession
from pyspark.sql import SparkSession

# Build the SparkSession
spark = SparkSession.builder \
   .master("local") \
   .appName("Linear Regression Model") \
   .config("spark.executor.memory", "1gb") \
   .getOrCreate()
   
sc = spark.sparkContext

Nota che se ottieni un errore FileNotFoundError simile a questo: “No such file or directory: ‘/User/YourName/Downloads/spark-2.1.0-bin-hadoop2.7/./bin/spark-submit’”, sai che devi (re)impostare il tuo PATH di Spark. Vai alla tua home con $ cd e poi modifica il file .bash_profile eseguendo $ nano .bash_profile.

Aggiungi qualcosa come il seguente in fondo al file

export SPARK_HOME="/usr/local/spark"

Usa CTRL + X per uscire dal file ma assicurati di salvare le modifiche premendo anche Y per confermare. Poi, non dimenticare di rendere effettive le modifiche eseguendo source .bash_profile.

Consiglio: puoi anche impostare variabili d’ambiente aggiuntive se vuoi; Probabilmente non ti servono, ma è bene sapere che puoi farlo se necessario. Considera i seguenti esempi:

# Set a fixed value for the hash seed secret
export PYTHONHASHSEED=0

# Set an alternate Python executable
export PYSPARK_PYTHON=/usr/local/ipython/bin/ipython

# Augment the default search path for shared libraries
export LD_LIBRARY_PATH=/usr/local/ipython/bin/ipython

# Augment the default search path for private libraries 
export PYTHONPATH=$SPARK_HOME/python/lib/py4j-*-src.zip:$PYTHONPATH:$SPARK_HOME/python/

Nota anche che ora hai inizializzato una SparkSession predefinita. Tuttavia, nella maggior parte dei casi vorrai configurarla ulteriormente. Vedrai che sarà davvero necessario quando inizierai a lavorare con big data. Se vuoi saperne di più, consulta questa pagina.

Caricare i tuoi dati

Questo tutorial utilizza il dataset California Housing. È apparso in un articolo del 1997, Sparse Spatial Autoregressions, scritto da Pace, R. Kelley e Ronald Barry e pubblicato su Statistics and Probability Letters. I ricercatori hanno costruito questo dataset utilizzando i dati del censimento della California del 1990.

I dati contengono una riga per ogni gruppo di blocchi censuari. Un block group è la più piccola unità geografica per cui l’U.S. Census Bureau pubblica dati campionari (un block group tipicamente ha una popolazione da 600 a 3.000 persone). In questo campione un block group include in media 1425,5 individui che vivono in un’area geograficamente compatta. Puoi ricavare queste informazioni da questa pagina web o leggendo l’articolo menzionato sopra, che trovi qui.

Questi dati spaziali contengono 20.640 osservazioni sui prezzi delle case con 9 variabili economiche:

• Longitude si riferisce alla distanza angolare di un luogo geografico a nord o a sud dell’equatore terrestre per ciascun block group;
• Latitude si riferisce alla distanza angolare di un luogo geografico a est o a ovest dell’equatore terrestre per ciascun block group;
• Housing median age è l’età mediana delle persone che appartengono a un block group. Nota che la mediana è il valore al punto medio di una distribuzione di frequenza dei valori osservati;
• Total rooms è il numero totale di stanze nelle case per block group;
• Total bedrooms è il numero totale di camere da letto nelle case per block group;
• Population è il numero di abitanti di un block group;
• Households si riferisce alle unità abitative e ai loro occupanti per block group;
• Median income registra il reddito mediano delle persone che appartengono a un block group; e
• Median house value è la variabile dipendente e si riferisce al valore mediano delle case per block group.

Inoltre, scopri che tutti i block group con zeri nelle variabili indipendenti e dipendenti sono stati esclusi dai dati.

Median house value è la variabile dipendente e avrà il ruolo di variabile target nel tuo modello di ML.

Puoi scaricare i dati qui. Cerca la cartella houses.zip, scaricala e scompattala per accedere alle cartelle dei dati.

Poi userai il metodo textFile() per leggere i dati dalla cartella in cui li hai scaricati in RDD. Questo metodo riceve un URI per il file, che in questo caso è il percorso locale della tua macchina, e lo legge come una collezione di righe. Per comodità, leggerai non solo il file .data, ma anche il file .domain che contiene l’header. Questo ti permetterà di verificare l’ordine delle variabili.

# Load in the data
rdd = sc.textFile('/Users/yourName/Downloads/CaliforniaHousing/cal_housing.data')

# Load in the header
header = sc.textFile('/Users/yourName/Downloads/CaliforniaHousing/cal_housing.domain')

Esplorazione dei dati

Hai già ricavato molte informazioni semplicemente guardando la pagina web in cui hai trovato il dataset, ma è sempre meglio sporcarsi le mani e ispezionare i dati con l’aiuto di Spark con Python, in questo caso.

È importante capire che, poiché l’esecuzione di Spark è “lazy”, non è stato ancora eseguito nulla. I tuoi dati non sono stati effettivamente letti. Le variabili rdd e header sono in realtà solo concetti nella tua mente. Devi spingere Spark a lavorare per te, quindi usiamo il metodo collect() per guardare header:

header.collect()

Il metodo collect() porta l’intero RDD su una singola macchina, e vedrai il seguente risultato:

[u'longitude: continuous.', u'latitude: continuous.', u'housingMedianAge: continuous. ', u'totalRooms: continuous. ', u'totalBedrooms: continuous. ', u'population: continuous. ', u'households: continuous. ', u'medianIncome: continuous. ', u'medianHouseValue: continuous. ']

Consiglio: fai attenzione quando usi collect()! Eseguire questa riga può far esaurire la memoria al driver. Per questo l’approccio seguente con il metodo take() è più sicuro se vuoi stampare solo alcuni elementi dell’RDD. In generale, è una buona pratica limitare il result set quando possibile, proprio come quando usi SQL.

Scopri che l’ordine delle variabili è lo stesso visto sopra nella presentazione del dataset, e anche che tutte le colonne dovrebbero avere valori continui. Forziamo Spark a fare un po’ più di lavoro e diamo un’occhiata ai dati sulle case in California per confermarlo.

Chiama il metodo take() sul tuo RDD:

rdd.take(2)

Eseguendo la riga precedente, prendi i primi 2 elementi dell’RDD. Il risultato è come previsto: poiché hai letto i file con la funzione textFile(), le righe sono state lette tutte insieme. Le voci sono separate da una singola virgola e le righe stesse sono separate anch’esse da una virgola:

[u'-122.230000,37.880000,41.000000,880.000000,129.000000,322.000000,126.000000,8.325200,452600.000000', u'-122.220000,37.860000,21.000000,7099.000000,1106.000000,2401.000000,1138.000000,8.301400,358500.000000']

Devi decisamente risolvere questo. Ora, non devi dividere le voci, ma devi sicuramente assicurarti che le righe dei tuoi dati siano elementi separati. Per risolvere, userai la funzione map() a cui passerai una funzione lambda per dividere la riga sulla virgola. Poi controlla il risultato eseguendo la stessa riga con il metodo take(), come fatto prima:

Ricorda che le funzioni lambda sono funzioni anonime create a runtime.

# Split lines on commas
rdd = rdd.map(lambda line: line.split(","))

# Inspect the first 2 lines 
rdd.take(2)

Otterrai il seguente risultato:

[[u'-122.230000', u'37.880000', u'41.000000', u'880.000000', u'129.000000', u'322.000000', u'126.000000', u'8.325200', u'452600.000000'], [u'-122.220000', u'37.860000', u'21.000000', u'7099.000000', u'1106.000000', u'2401.000000', u'1138.000000', u'8.301400', u'358500.000000']]

In alternativa, puoi anche usare le seguenti funzioni per ispezionare i dati:

# Inspect the first line 
rdd.first()

# Take top elements
rdd.top(2)

Se sei abituato a lavorare con Pandas o con i data frame in R, probabilmente ti aspetteresti anche un header, ma qui non c’è. Per semplificarti la vita, passerai dagli RDD a un DataFrame. I DataFrame sono preferibili agli RDD quando puoi usarli. Soprattutto in Python, le prestazioni dei DataFrame sono migliori degli RDD.

Ma qual è la differenza tra i due?

Puoi usare gli RDD quando vuoi eseguire trasformazioni e azioni di basso livello sui tuoi dati non strutturati. Significa che non ti interessa imporre uno schema durante l’elaborazione o accedere agli attributi per nome o colonna. Collegandoci a quanto detto sulle prestazioni, usando gli RDD non cerchi necessariamente i benefici prestazionali che i DataFrame possono offrire per dati (semi-)strutturati. Usa gli RDD quando vuoi manipolare i dati con costrutti di programmazione funzionale piuttosto che espressioni specifiche di dominio.

Per ricapitolare, passerai ora ai DataFrame per usare espressioni di alto livello, eseguire query SQL per esplorare ulteriormente i dati e ottenere accesso colonnare.

Quindi facciamolo.

Il primo passo è creare una SchemaRDD o un RDD di oggetti Row con uno schema. È normale, perché proprio come per un DataFrame, vuoi arrivare a una situazione con righe e colonne. Ogni voce è collegata a una riga e a una certa colonna e le colonne hanno tipi di dato.

Userai ancora la funzione map() e un’altra funzione lambda in cui mapperai ogni voce a un campo in una Row. Per visualizzarlo, considera questa prima riga:

[u'-122.230000', u'37.880000', u'41.000000', u'880.000000', u'129.000000', u'322.000000', u'126.000000', u'8.325200', u'452600.000000']

La funzione lambda dice che costruirai una riga in una SchemaRDD e che l’elemento all’indice 0 avrà il nome “longitude”, e così via.

Con questa SchemaRDD in atto, puoi facilmente convertire l’RDD in un DataFrame con il metodo toDF().

# Import the necessary modules 
from pyspark.sql import Row

# Map the RDD to a DF
df = rdd.map(lambda line: Row(longitude=line[0], 
                              latitude=line[1], 
                              housingMedianAge=line[2],
                              totalRooms=line[3],
                              totalBedRooms=line[4],
                              population=line[5], 
                              households=line[6],
                              medianIncome=line[7],
                              medianHouseValue=line[8])).toDF()

Ora che hai il tuo DataFrame df, puoi ispezionarlo con i metodi già usati prima, cioè first() e take(), ma anche con head() e show():

# Show the top 20 rows 
df.show()

Vedrai subito che è molto diverso dall’RDD con cui stavi lavorando prima:

Consiglio: usa df.columns per restituire le colonne del tuo DataFrame.

I dati sembrano ordinati in colonne, ma che dire dei tipi di dato? Leggendo i dati, Spark proverà a inferire uno schema, ma qui c’è riuscito? Usa df.dtypes o df.printSchema() per conoscere meglio i tipi di dato contenuti nel tuo DataFrame.

# Print the data types of all `df` columns
# df.dtypes

# Print the schema of `df`
df.printSchema()

Poiché non esegui la prima riga di codice, otterrai solo il seguente risultato:

root
 |-- households: string (nullable = true)
 |-- housingMedianAge: string (nullable = true)
 |-- latitude: string (nullable = true)
 |-- longitude: string (nullable = true)
 |-- medianHouseValue: string (nullable = true)
 |-- medianIncome: string (nullable = true)
 |-- population: string (nullable = true)
 |-- totalBedRooms: string (nullable = true)
 |-- totalRooms: string (nullable = true)

Tutte le colonne sono ancora di tipo string… Che delusione!

Se vuoi continuare con questo DataFrame, dovrai rimediare e assegnare tipi di dato “migliori” o più accurati a tutte le colonne. Anche le prestazioni ne beneficeranno. Intuitivamente, potresti optare per una soluzione come la seguente, in cui dichiari che ogni colonna del DataFrame df deve essere convertita in FloatType():

from pyspark.sql.types import *

df = df.withColumn("longitude", df["longitude"].cast(FloatType())) \
   .withColumn("latitude", df["latitude"].cast(FloatType())) \
   .withColumn("housingMedianAge",df["housingMedianAge"].cast(FloatType())) \
   .withColumn("totalRooms", df["totalRooms"].cast(FloatType())) \ 
   .withColumn("totalBedRooms", df["totalBedRooms"].cast(FloatType())) \ 
   .withColumn("population", df["population"].cast(FloatType())) \ 
   .withColumn("households", df["households"].cast(FloatType())) \ 
   .withColumn("medianIncome", df["medianIncome"].cast(FloatType())) \ 
   .withColumn("medianHouseValue", df["medianHouseValue"].cast(FloatType()))

Ma queste chiamate ripetute sono piuttosto oscure, soggette a errori e non molto eleganti. Perché non scrivere una funzione che faccia tutto questo in modo più pulito?

La seguente User-Defined Function (UDF) prende un DataFrame, i nomi delle colonne e il nuovo tipo di dato che vuoi assegnare. Dici che per ogni nome di colonna prendi la colonna e la converti al nuovo tipo. Poi restituisci il DataFrame:

# Import all from `sql.types`
from pyspark.sql.types import *

# Write a custom function to convert the data type of DataFrame columns
def convertColumn(df, names, newType):
  for name in names: 
     df = df.withColumn(name, df[name].cast(newType))
  return df 

# Assign all column names to `columns`
columns = ['households', 'housingMedianAge', 'latitude', 'longitude', 'medianHouseValue', 'medianIncome', 'population', 'totalBedRooms', 'totalRooms']

# Conver the `df` columns to `FloatType()`
df = convertColumn(df, columns, FloatType())

Molto meglio! Puoi ispezionare rapidamente i tipi di dato di df con il metodo printSchema(), come fatto prima.

Ora che hai sistemato questo, è ora di iniziare davvero con l’esplorazione dei dati. Hai visto che accesso colonnare e query SQL erano due vantaggi dei DataFrame. Bene, ora approfondiamo. Iniziamo in piccolo e selezioniamo due colonne da df di cui vuoi vedere solo 10 righe:

df.select('population','totalBedRooms').show(10)

Questa query ti dà il seguente risultato:

+----------+-------------+
|population|totalBedRooms|
+----------+-------------+
|     322.0|        129.0|
|    2401.0|       1106.0|
|     496.0|        190.0|
|     558.0|        235.0|
|     565.0|        280.0|
|     413.0|        213.0|
|    1094.0|        489.0|
|    1157.0|        687.0|
|    1206.0|        665.0|
|    1551.0|        707.0|
+----------+-------------+
only showing top 10 rows

Puoi anche rendere le query più complesse, come vedi nel seguente esempio:

df.groupBy("housingMedianAge").count().sort("housingMedianAge",ascending=False).show()

Che ti dà il seguente risultato:

+----------------+-----+                                                        
|housingMedianAge|count|
+----------------+-----+
|            52.0| 1273|
|            51.0|   48|
|            50.0|  136|
|            49.0|  134|
|            48.0|  177|
|            47.0|  198|
|            46.0|  245|
|            45.0|  294|
|            44.0|  356|
|            43.0|  353|
|            42.0|  368|
|            41.0|  296|
|            40.0|  304|
|            39.0|  369|
|            38.0|  394|
|            37.0|  537|
|            36.0|  862|
|            35.0|  824|
|            34.0|  689|
|            33.0|  615|
+----------------+-----+
only showing top 20 rows

Oltre alle query, puoi anche scegliere di descrivere i tuoi dati e ottenere statistiche riassuntive. Questo ti aiuterà sicuramente dopo!

df.describe().show()

Guarda i valori minimi e massimi di tutti gli attributi (numerici). Noti che più attributi hanno un ampio intervallo di valori: dovrai normalizzare il dataset.

Preprocessing dei dati

Con tutte le informazioni raccolte dalla tua piccola analisi esplorativa, sai abbastanza per pre-processare i dati da dare in pasto al modello.

• Non dovresti preoccuparti dei valori mancanti; tutti i valori zero sono stati esclusi dal dataset.
• Probabilmente dovresti standardizzare i dati, visto che l’intervallo dei valori minimi e massimi è piuttosto ampio.
• Potrebbero esserci ulteriori attributi che potresti aggiungere, come una feature che registra il numero di camere da letto per stanza o le stanze per nucleo familiare.
• Anche la tua variabile dipendente è piuttosto grande; Per semplificarti la vita, dovrai leggermente aggiustarne i valori.

# Import all from `sql.functions` 
from pyspark.sql.functions import *

# Adjust the values of `medianHouseValue`
df = df.withColumn("medianHouseValue", col("medianHouseValue")/100000)

# Show the first 2 lines of `df`
df.take(2)

[Row(households=126.0, housingMedianAge=41.0, latitude=37.880001068115234, longitude=-122.2300033569336, medianHouseValue=4.526, medianIncome=8.325200080871582, population=322.0, totalBedRooms=129.0, totalRooms=880.0), Row(households=1138.0, housingMedianAge=21.0, latitude=37.86000061035156, longitude=-122.22000122070312, medianHouseValue=3.585, medianIncome=8.301400184631348, population=2401.0, totalBedRooms=1106.0, totalRooms=7099.0)]

# Import all from `sql.functions` if you haven't yet
from pyspark.sql.functions import *

# Divide `totalRooms` by `households`
roomsPerHousehold = df.select(col("totalRooms")/col("households"))

# Divide `population` by `households`
populationPerHousehold = df.select(col("population")/col("households"))

# Divide `totalBedRooms` by `totalRooms`
bedroomsPerRoom = df.select(col("totalBedRooms")/col("totalRooms"))

# Add the new columns to `df`
df = df.withColumn("roomsPerHousehold", col("totalRooms")/col("households")) \
   .withColumn("populationPerHousehold", col("population")/col("households")) \
   .withColumn("bedroomsPerRoom", col("totalBedRooms")/col("totalRooms"))
   
# Inspect the result
df.first()

Row(households=126.0, housingMedianAge=41.0, latitude=37.880001068115234, longitude=-122.2300033569336, medianHouseValue=4.526, medianIncome=8.325200080871582, population=322.0, totalBedRooms=129.0, totalRooms=880.0, roomsPerHousehold=6.984126984126984, populationPerHousehold=2.5555555555555554, bedroomsPerRoom=0.14659090909090908)

# Re-order and select columns
df = df.select("medianHouseValue", 
              "totalBedRooms", 
              "population", 
              "households", 
              "medianIncome", 
              "roomsPerHousehold", 
              "populationPerHousehold", 
              "bedroomsPerRoom")

# Import `DenseVector`
from pyspark.ml.linalg import DenseVector

# Define the `input_data` 
input_data = df.rdd.map(lambda x: (x[0], DenseVector(x[1:])))

# Replace `df` with the new DataFrame
df = spark.createDataFrame(input_data, ["label", "features"])

# Import `StandardScaler` 
from pyspark.ml.feature import StandardScaler

# Initialize the `standardScaler`
standardScaler = StandardScaler(inputCol="features", outputCol="features_scaled")

# Fit the DataFrame to the scaler
scaler = standardScaler.fit(df)

# Transform the data in `df` with the scaler
scaled_df = scaler.transform(df)

# Inspect the result
scaled_df.take(2)

[Row(label=4.526, features=DenseVector([129.0, 322.0, 126.0, 8.3252, 6.9841, 2.5556, 0.1466]), features_scaled=DenseVector([0.3062, 0.2843, 0.3296, 4.3821, 2.8228, 0.2461, 2.5264])), Row(label=3.585, features=DenseVector([1106.0, 2401.0, 1138.0, 8.3014, 6.2381, 2.1098, 0.1558]), features_scaled=DenseVector([2.6255, 2.1202, 2.9765, 4.3696, 2.5213, 0.2031, 2.6851]))]

# Split the data into train and test sets
train_data, test_data = scaled_df.randomSplit([.8,.2],seed=1234)

# Import `LinearRegression`
from pyspark.ml.regression import LinearRegression

# Initialize `lr`
lr = LinearRegression(labelCol="label", maxIter=10, regParam=0.3, elasticNetParam=0.8)

# Fit the data to the model
linearModel = lr.fit(train_data)

# Generate predictions
predicted = linearModel.transform(test_data)

# Extract the predictions and the "known" correct labels
predictions = predicted.select("prediction").rdd.map(lambda x: x[0])
labels = predicted.select("label").rdd.map(lambda x: x[0])

# Zip `predictions` and `labels` into a list
predictionAndLabel = predictions.zip(labels).collect()

# Print out first 5 instances of `predictionAndLabel` 
predictionAndLabel[:5]

[(1.4491508524918457, 0.14999), (1.5705029404692372, 0.14999), (2.148727956912464, 0.14999), (1.5831547768979277, 0.344), (1.5182107797955968, 0.398)]

# Coefficients for the model
linearModel.coefficients

# Intercept for the model
linearModel.intercept

# The coefficients
[0.0,0.0,0.0,0.276239709215,0.0,0.0,0.0]

# The intercept
0.990399577462

# Get the RMSE
linearModel.summary.rootMeanSquaredError

# Get the R2
linearModel.summary.r2

# RMSE
0.8692118678997669

# R2
0.4240895287218379

spark.stop()