Apache Spark-tutorial: ML met PySpark

pyspark downloaden met pip

Vervolgens kun je PySpark downloaden en installeren met behulp van pip. Dit is vrij eenvoudig en lijkt op het installeren van elk ander pakket. Je voert gewoon het gebruikelijke commando uit en het zware werk wordt voor je gedaan:

$ pip install pyspark

Als alternatief kun je ook naar de Spark-downloadpagina gaan. Laat de standaardopties in de eerste drie stappen staan en je vindt in stap 4 een downloadlink. Klik op die link om te downloaden. Voor deze tutorial download je de Spark Release 2.2.0 en het pakkettype “Pre-built for Apache Hadoop 2.7 and later”.

Let op dat het downloaden even kan duren!

Spark downloaden met Homebrew

Je kunt Spark ook installeren met Homebrew, een gratis en opensource-pakketbeheerder. Dit is vooral handig als je met macOS werkt.

Voer simpelweg de volgende commando's uit om naar Spark te zoeken, meer informatie op te vragen en het uiteindelijk op je computer te installeren:

# Search for spark
$ brew search spark

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

# Install apache-spark
$ brew install apache-spark

Spark downloaden en instellen

Zorg er vervolgens voor dat je de map uitpakt die in je Downloads-map verschijnt. Dit kan automatisch gebeuren door dubbel te klikken op het archief spark-2.2.0-bin-hadoop2.7.tgz of door je Terminal te openen en het volgende commando uit te voeren:

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

Verplaats daarna de uitgepakte map naar /usr/local/spark met de volgende regel:

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

Let op dat als je een foutmelding krijgt dat de machtiging is geweigerd om deze map te verplaatsen, je sudo voor dit commando moet zetten. De regel hierboven wordt dan $ sudo mv spark-2.1.0-bin-hadoop2.7 /usr/local/spark. Je wordt om je wachtwoord gevraagd, meestal hetzelfde als waarmee je je pc ontgrendelt bij het opstarten :)

Nu je klaar bent, open je het README-bestand op het pad /usr/local/spark. Dit doe je door uit te voeren

$ cd /usr/local/spark

Dit brengt je naar de map waar je moet zijn. Daarna kun je de map bekijken en het README-bestand lezen dat erbij hoort.

Gebruik eerst $ ls om een lijst te krijgen van de bestanden en mappen in deze spark-map. Je ziet dat er een README.md-bestand in staat. Je kunt het openen met een van de volgende commando's:

# Open and edit the file
$ nano README.md

# Just read the file 
$ cat README.md

Tip gebruik de tab-toets op je toetsenbord om automatisch aan te vullen terwijl je de bestandsnaam typt :) Dat scheelt tijd.

Je ziet dat deze README je algemene informatie geeft over Spark, online documentatie, Spark bouwen, de interactieve Scala- en Python-shells, voorbeeldprogramma's en meer.

Het meest interessante hier is misschien de sectie over het bouwen van Spark, maar let op: dit is vooral relevant als je geen prebuilt-versie hebt gedownload. Voor deze tutorial heb je echter een prebuilt-versie gedownload. Je kunt op CTRL + X drukken om de README te sluiten en terug te keren naar de spark-map.

Als je een versie hebt gekozen die nog niet is gebouwd, zorg dan dat je het commando uitvoert dat in de README staat. Ten tijde van schrijven is dat het volgende:

$ build/mvn -DskipTests clean package run

Let op dat dit commando even kan duren.

Spark-applicaties versus Spark-shell

De interactieve shell is een voorbeeld van een Read-Eval(uate)-Print-Loop (REPL)-omgeving; Dat betekent dat alles wat je typt wordt gelezen, geëvalueerd en teruggeprint zodat je je analyse kunt voortzetten. Dit doet denken aan IPython, een krachtige interactieve Python-shell die je misschien kent van Jupyter. Wil je meer weten, lees dan DataCamp’s blogpost IPython or Jupyter.

Dit betekent dat je de shell, die beschikbaar is voor zowel Python als Scala, kunt gebruiken voor al het interactieve werk dat je moet doen.

Naast deze shell kun je ook Spark-applicaties schrijven en deployen. In tegenstelling tot het schrijven van Spark-applicaties is in de shell de SparkSession al voor je aangemaakt, zodat je meteen aan de slag kunt zonder tijd te verliezen aan het creëren ervan.

Je vraagt je nu misschien af: wat is de SparkSession?

Dat is het hoofdingangspunt voor Spark-functionaliteit: het stelt de verbinding met een Spark-cluster voor en je gebruikt het om RDD's te maken en variabelen op dat cluster te broadcasten. Als je met Spark werkt, begint en eindigt alles met deze SparkSession. Let op dat vóór Spark 2.0.0 de drie belangrijkste verbindingsobjecten SparkContext, SqlContext en HiveContext waren.

Je ziet hier later meer van. Voor nu focussen we op de shell.

De Python Spark-shell

Vanuit de spark-map op /usr/local/spark kun je uitvoeren

$ ./bin/pyspark

Eerst verschijnt er wat tekst. Daarna zie je “Spark” verschijnen, zoals dit:

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'.
>>>

Als je dit ziet, weet je dat je klaar bent om te experimenteren in de interactieve shell!

Tip: als je liever de IPython-shell gebruikt dan de Spark-shell, kan dat door de volgende omgevingsvariabele te zetten:

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

Voor dit deel van de tutorial gebruik je niet de ishell maar bouw je je eigen applicatie. Dat doe je in een Jupyter Notebook. Je hebt alles al geïnstalleerd wat je nodig hebt, dus je hoeft weinig te doen om PySpark in Jupyter te laten werken.

Start de notebook-applicatie zoals altijd door $ jupyter notebook te draaien. Maak dan een nieuw notebook en importeer gewoon de findspark-bibliotheek en gebruik de functie init(). In dit geval geef je het pad /usr/local/spark mee aan init() omdat je zeker weet dat je Spark daar hebt geïnstalleerd.

# Import findspark 
import findspark

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

# Or use this alternative
#findspark.init()

Tip: als je geen idee hebt of je pad correct staat of waar je Spark hebt geïnstalleerd, kun je altijd findspark.find() gebruiken om de locatie automatisch te detecteren.

Zoek je alternatieve manieren om met Spark in Jupyter te werken, raadpleeg dan onze Apache Spark in Python: Beginner’s Guide.

Nu dat geregeld is, kun je eindelijk je eerste Spark-programma maken!

Je eerste Spark-programma maken

Wat je eerst wilt doen is de SparkContext importeren uit het pyspark-pakket en initialiseren. Onthoud dat je dit eerder niet hoefde te doen omdat de interactieve Spark-shell dit automatisch voor je aanmaakte en initialiseerde! Hier moet je zelf iets meer doen :)

Importeer de module SparkSession uit pyspark.sql en bouw een SparkSession met de methode builder(). Vervolgens kun je de master-URL instellen om mee te verbinden, de applicatienaam, extra configuratie toevoegen zoals het geheugen van de executor en tot slot getOrCreate() gebruiken om de huidige Spark-sessie op te halen of er een te maken als er geen draait.

# 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

Let op dat als je een FileNotFoundError krijgt zoals: “No such file or directory: ‘/User/YourName/Downloads/spark-2.1.0-bin-hadoop2.7/./bin/spark-submit’”, je weet dat je je Spark PATH moet (her)instellen. Ga naar je home directory met $ cd en bewerk dan het bestand .bash_profile met $ nano .bash_profile.

Voeg onderaan zoiets toe als

export SPARK_HOME="/usr/local/spark"

Gebruik CTRL + X om het bestand te sluiten, maar zorg dat je je aanpassingen bewaart door Y in te voeren om de wijzigingen te bevestigen. Vergeet daarna niet de wijzigingen te activeren met source .bash_profile.

Tip: je kunt ook extra omgevingsvariabelen instellen als je wilt; Waarschijnlijk heb je ze niet nodig, maar het is goed om te weten dat het kan. Overweeg de volgende voorbeelden:

# 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/

Let op ook dat je nu een standaard SparkSession hebt geïnitialiseerd. In de meeste gevallen wil je deze verder configureren. Je ziet dat dit echt nodig wordt zodra je met big data werkt. Wil je hier meer over weten, bekijk dan deze pagina.

Je data inladen

Deze tutorial gebruikt de California Housing-dataset. Die verscheen in een paper uit 1997 getiteld Sparse Spatial Autoregressions, geschreven door Pace, R. Kelley en Ronald Barry en gepubliceerd in Statistics and Probability Letters. De onderzoekers bouwden deze dataset met gegevens uit de volkstelling van Californië van 1990.

De data bevat één rij per block group van de volkstelling. Een block group is de kleinste geografische eenheid waarvoor het U.S. Census Bureau steekproefdata publiceert (een block group heeft doorgaans 600 tot 3.000 inwoners). In deze steekproef omvat een block group gemiddeld 1425,5 personen die in een geografisch compacte zone wonen. Je haalt deze informatie van deze webpagina of door het bovengenoemde paper te lezen, dat je hier vindt.

Deze ruimtelijke data bevatten 20.640 observaties over huizenprijzen met 9 economische variabelen:

• Longitude verwijst naar de hoekafstand van een geografische plaats ten noorden of zuiden van de evenaar van de aarde per block group;
• Latitude verwijst naar de hoekafstand van een geografische plaats ten oosten of westen van de evenaar van de aarde per block group;
• Housing median age is de mediane leeftijd van de mensen die tot een block group behoren. Let op dat de mediaan de waarde is die in het midden ligt van een frequentieverdeling van geobserveerde waarden;
• Total rooms is het totale aantal kamers in de huizen per block group;
• Total bedrooms is het totale aantal slaapkamers in de huizen per block group;
• Population is het aantal inwoners van een block group;
• Households verwijst naar huishoudens en hun bewoners per block group;
• Median income registreert het mediaaninkomen van mensen die tot een block group behoren; En,
• Median house value is de afhankelijke variabele en verwijst naar de mediane woningwaarde per block group.

Bovendien leer je dat alle block groups met nullen voor de onafhankelijke en afhankelijke variabelen uit de data zijn verwijderd.

De Median house value is de afhankelijke variabele en krijgt de rol van doelvariabele in je ML-model.

Je kunt de data hier downloaden. Zoek de map houses.zip, download en pak uit zodat je toegang hebt tot de datamappen.

Vervolgens gebruik je de methode textFile() om de data vanaf de map waar je deze hebt gedownload in te lezen naar RDD's. Deze methode neemt een URI van het bestand, in dit geval het lokale pad van je machine, en leest dit als een collectie regels. Voor het gemak lees je niet alleen het .data-bestand in, maar ook het .domain-bestand met de header. Zo kun je de volgorde van de variabelen dubbelchecken.

# 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')

Data verkennen

Je hebt al veel informatie verzameld door simpelweg naar de webpagina van de dataset te kijken, maar het is altijd beter om hands-on je data te inspecteren met Spark met Python, in dit geval.

Belangrijk om te begrijpen is dat, omdat Spark “lazy” uitvoert, er nog niets is uitgevoerd. Je data is nog niet echt ingelezen. De variabelen rdd en header zijn eigenlijk alleen concepten in je hoofd. Je moet Spark aan het werk zetten, dus gebruik de methode collect() om naar de header te kijken:

header.collect()

De methode collect() brengt de volledige RDD naar één machine, en je ziet het volgende resultaat:

[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. ']

Tip: wees voorzichtig met collect()! Deze regel kan ervoor zorgen dat het geheugen van de driver volloopt. De volgende aanpak met de methode take() is veiliger als je slechts een paar elementen van de RDD wilt printen. In het algemeen is het een goed principe om je resultset te beperken waar mogelijk, net als bij SQL.

Je leert dat de volgorde van de variabelen overeenkomt met wat je eerder zag bij de presentatie van de dataset, en dat alle kolommen continue waarden zouden moeten hebben. Laten we Spark wat meer werk laten doen en naar de California housing-data kijken om dit te bevestigen.

Roep de methode take() aan op je RDD:

rdd.take(2)

Door de vorige regel code uit te voeren neem je de eerste 2 elementen van de RDD. Het resultaat is zoals verwacht: omdat je de bestanden met de functie textFile() hebt ingelezen, zijn de regels gewoon allemaal als tekst ingelezen. De waarden zijn gescheiden door een komma en de rijen zelf ook door een komma:

[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']

Dit moet je zeker oplossen. Je hoeft de waarden niet nu al te splitsen, maar je moet er wel voor zorgen dat de rijen van je data losse elementen zijn. Om dit op te lossen, gebruik je de functie map() waaraan je een lambda-functie meegeeft om de regel op de komma te splitsen. Controleer daarna je resultaat met take(), net als daarnet:

Onthoud dat lambda-functies anonieme functies zijn die tijdens runtime worden aangemaakt.

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

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

Je krijgt het volgende resultaat:

[[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']]

Als alternatief kun je ook de volgende functies gebruiken om je data te inspecteren:

# Inspect the first line 
rdd.first()

# Take top elements
rdd.top(2)

Als je gewend bent met Pandas of data frames in R te werken, had je waarschijnlijk ook een header verwacht, maar die is er niet. Om het jezelf makkelijker te maken, stap je over van de RDD naar een DataFrame. DataFrames hebben de voorkeur boven RDD's wanneer je ze kunt gebruiken. Zeker met Python is de performance van DataFrames beter dan die van RDD's.

Maar wat is het verschil tussen die twee?

Je gebruikt RDD's wanneer je low-level transformaties en acties op ongestructureerde data wilt uitvoeren. Dat betekent dat je geen schema wilt opleggen tijdens de verwerking of attributen op naam/kolom wilt benaderen. Aansluitend op wat eerder over performance werd gezegd: met RDD's mik je niet per se op de prestatievoordelen die DataFrames bieden voor (semi)gestructureerde data. Gebruik RDD's wanneer je de data wilt manipuleren met functionele programmeerconstructies in plaats van domeinspecifieke expressies.

Samengevat stap je nu over op DataFrames om high-level expressies te gebruiken, SQL-queries uit te voeren om je data verder te verkennen en kolomtoegang te krijgen.

Laten we dat doen.

De eerste stap is een SchemaRDD of een RDD van Row-objecten met een schema maken. Dat is logisch, want net als bij een DataFrame wil je uiteindelijk rijen en kolommen hebben. Elke entry is gekoppeld aan een rij en een bepaalde kolom en kolommen hebben datatypes.

Je gebruikt opnieuw de functie map() en een lambda-functie waarin je elke entry naar een veld in een Row mapt. Om dit visueler te maken, bekijk deze eerste regel:

[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']

De lambda-functie zegt dat je een rij in een SchemaRDD gaat construeren en dat het element op index 0 de naam “longitude” krijgt, enzovoort.

Met deze SchemaRDD kun je de RDD eenvoudig omzetten naar een DataFrame met de methode 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()

Nu je je DataFrame df hebt, kun je het inspecteren met de methoden die je eerder ook gebruikte, namelijk first() en take(), maar ook met head() en show():

# Show the top 20 rows 
df.show()

Je ziet meteen dat dit er heel anders uitziet dan de RDD waarmee je eerder werkte:

Tip: gebruik df.columns om de kolommen van je DataFrame op te vragen.

De data lijken netjes in kolommen geordend, maar hoe zit het met de datatypes? Bij het inlezen probeert Spark een schema af te leiden, maar is dat hier gelukt? Gebruik df.dtypes of df.printSchema() om meer te weten te komen over de datatypes in je DataFrame.

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

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

Omdat je de eerste regel niet uitvoert, krijg je alleen het volgende resultaat terug:

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)

Alle kolommen zijn nog steeds van het type string… Dat is teleurstellend!

Als je met dit DataFrame wilt doorgaan, moet je dit rechtzetten en “betere” of nauwkeurigere datatypes aan alle kolommen toekennen. Je performance profiteert hier ook van. Intuïtief kun je kiezen voor een oplossing als de volgende, waarin je opgeeft dat elke kolom van DataFrame df naar een FloatType() moet worden gecast:

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()))

Maar deze herhaalde aanroepen zijn nogal ondoorzichtig, foutgevoelig en ogen niet mooi. Waarom schrijf je niet een functie die dit schoner voor je doet?

De volgende user-defined function (UDF) neemt een DataFrame, kolomnamen en het nieuwe datatype dat je de kolommen wilt geven. Je zegt dat je voor elke kolomnaam de kolom pakt en naar een nieuw datatype cast. Vervolgens retourneer je het 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())

Dat ziet er al veel beter uit! Je kunt de datatypes van df snel inspecteren met de methode printSchema(), net als eerder.

Nu dit allemaal geregeld is, is het tijd om echt te beginnen met de data-exploratie. Je hebt gezien dat kolomtoegang en SQL-queries twee voordelen van DataFrames zijn. Tijd om daar dieper in te duiken. Begin klein en selecteer twee kolommen uit df waarvan je slechts 10 rijen wilt zien:

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

Deze query geeft je het volgende resultaat:

+----------+-------------+
|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

Je kunt je queries ook complexer maken, zoals in het volgende voorbeeld:

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

Wat je het volgende resultaat geeft:

+----------------+-----+                                                        
|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

Naast query'en kun je er ook voor kiezen je data te beschrijven en enkele samenvattende statistieken op te vragen. Dit helpt je zeker verderop!

df.describe().show()

Kijk naar de minimum- en maximumwaarden van alle (numerieke) attributen. Je ziet dat meerdere attributen een brede range hebben: je zult je dataset moeten normaliseren.

Datapreprocessing

Met alle informatie uit je kleine verkennende analyse weet je genoeg om je data te preprocessen en aan het model te voeren.

• Je hoeft je geen zorgen te maken over missende waarden; alle nullen zijn uit de dataset verwijderd.
• Je zou je data waarschijnlijk moeten standaardiseren, zoals je zag dat de range van minimum- en maximumwaarden vrij groot is.
• Er zijn mogelijk extra attributen toe te voegen, zoals een feature die het aantal slaapkamers per kamer of kamers per huishouden registreert.
• Je afhankelijke variabele is ook vrij groot; Om het jezelf makkelijker te maken, moet je de waarden licht aanpassen.

# 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()