Apache-Spark-Tutorial: ML mit PySpark

Herunterladen von pyspark mit pip

Du kannst PySpark ganz einfach mit pip herunterladen und installieren. Das funktioniert wie bei jedem anderen Paket: Du führst den bekannten Befehl aus und der Rest passiert automatisch:

$ pip install pyspark

Alternativ kannst du auch die Spark-Download-Seite nutzen. Lass die Standardoptionen in den ersten drei Schritten unverändert; in Schritt 4 findest du dann einen Download-Link. Klicke darauf, um den Download zu starten. Für dieses Tutorial lädst du Spark in der Version 2.2.0 und den Pakettyp „Pre-built for Apache Hadoop 2.7 and later“ herunter.

Hinweis: Der Download kann etwas dauern!

Spark mit Homebrew installieren

Du kannst Spark auch mit Homebrew, einem freien Open-Source-Paketmanager, installieren. Das ist besonders praktisch, wenn du mit macOS arbeitest.

Führe einfach die folgenden Befehle aus, um nach Spark zu suchen, mehr Infos zu erhalten und es schließlich auf deinem Rechner zu installieren:

# Search for spark
$ brew search spark

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

# Install apache-spark
$ brew install apache-spark

Spark herunterladen und einrichten

Entpacke nun das Archiv, das in deinem Ordner Downloads auftaucht. Das geht automatisch per Doppelklick auf spark-2.2.0-bin-hadoop2.7.tgz oder über das Terminal mit folgendem Befehl:

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

Verschiebe anschließend den entpackten Ordner nach /usr/local/spark mit dieser Zeile:

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

Hinweis: Wenn du eine Fehlermeldung zur fehlenden Berechtigung bekommst, füge sudo vor dem Befehl ein. Die Zeile lautet dann $ sudo mv spark-2.1.0-bin-hadoop2.7 /usr/local/spark. Du wirst nach deinem Passwort gefragt, in der Regel das gleiche, mit dem du deinen PC entsperrst :)

Jetzt öffne die README-Datei im Pfad /usr/local/spark. Das machst du mit

$ cd /usr/local/spark

Damit wechselst du in den benötigten Ordner. Dort kannst du dir die Dateien ansehen und die enthaltene README lesen.

Nutze zuerst $ ls, um dir Dateien und Ordner im spark-Verzeichnis auflisten zu lassen. Du siehst die Datei README.md. Öffne sie mit einem der folgenden Befehle:

# Open and edit the file
$ nano README.md

# Just read the file 
$ cat README.md

Tipp: Nutze die Tabulatortaste, um Dateinamen beim Tippen automatisch zu vervollständigen :) Das spart Zeit.

Die README enthält allgemeine Infos zu Spark, Online-Dokumentation, Build-Hinweise, die interaktiven Scala- und Python-Shells, Beispielprogramme und mehr.

Besonders interessant ist der Abschnitt zum Bauen von Spark – aber nur, wenn du keine vorgebaute Version geladen hast. Für dieses Tutorial hast du eine vorgebaute Version heruntergeladen. Beende die README mit STRG + X, um zurück in den Spark-Ordner zu gelangen.

Falls du doch eine Version gewählt hast, die noch nicht gebaut wurde, führe den in der README genannten Befehl aus. Zum Zeitpunkt der Erstellung war das:

$ build/mvn -DskipTests clean package run

Beachte, dass dieser Befehl eine Weile laufen kann.

Spark-Anwendungen vs. Spark-Shell

Die interaktive Shell ist eine Read-Eval(uate)-Print-Loop-Umgebung (REPL); Das heißt, Eingaben werden gelesen, ausgewertet und das Ergebnis ausgegeben, damit du direkt weitermachen kannst. Das erinnert an IPython, eine leistungsfähige interaktive Python-Shell, die du aus Jupyter kennst. Mehr dazu im DataCamp-Artikel IPython or Jupyter.

Du kannst die Shell, die es für Python und Scala gibt, für alle interaktiven Aufgaben nutzen.

Neben der Shell kannst du auch Spark-Anwendungen schreiben und deployen. Im Gegensatz dazu ist in der Shell die SparkSession bereits erstellt, sodass du direkt loslegen kannst, ohne Zeit für die Initialisierung zu verlieren.

Du fragst dich: Was ist die SparkSession?

Sie ist der zentrale Einstiegspunkt in die Spark-Funktionalität: Sie stellt die Verbindung zu einem Spark-Cluster dar, mit der du RDDs erzeugst und Variablen im Cluster broadcastest. In Spark beginnt und endet alles mit dieser SparkSession. Hinweis: Vor Spark 2.0.0 waren die drei zentralen Objekte SparkContext, SqlContext und HiveContext.

Mehr dazu später. Konzentrieren wir uns zunächst auf die Shell.

Die Python-Spark-Shell

Im Ordner spark unter /usr/local/spark startest du mit

$ ./bin/pyspark

Erst erscheinen einige Ausgaben, dann siehst du „Spark“, etwa so:

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

Wenn du diese Ausgabe siehst, kannst du in der interaktiven Shell loslegen!

Tipp: Falls du lieber die IPython-Shell statt der Spark-Shell nutzt, setze diese Umgebungsvariable:

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 im Jupyter Notebook

Für diesen Teil nutzt du nicht die Shell, sondern baust eine eigene Anwendung – in einem Jupyter Notebook. Alles Nötige ist bereits installiert, daher ist keine große Einrichtung erforderlich.

Starte das Notebook wie gewohnt mit $ jupyter notebook. Erstelle ein neues Notebook, importiere die Bibliothek findspark und rufe die Funktion init() auf. In diesem Beispiel gibst du /usr/local/spark an, weil du sicher bist, dass dort deine Spark-Installation liegt.

# Import findspark 
import findspark

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

# Or use this alternative
#findspark.init()

Tipp: Wenn du unsicher bist, ob der Pfad stimmt oder wo Spark installiert ist, nutze findspark.find(), um den Installationsort automatisch zu erkennen.

Weitere Wege, Spark in Jupyter zu nutzen, findest du in unserer Apache Spark in Python: Beginner’s Guide.

Jetzt kannst du endlich dein erstes Spark-Programm erstellen!

Dein erstes Spark-Programm

Als Erstes importierst du den SparkContext aus dem Paket pyspark und initialisierst ihn. Denke daran, dass die interaktive Spark-Shell dies automatisch für dich erledigt hat! Hier musst du etwas mehr selbst tun :)

Importiere das Modul SparkSession aus pyspark.sql und baue mit der Methode builder() eine SparkSession. Danach setzt du die Master-URL, den Applikationsnamen, optional zusätzliche Konfiguration wie den Executor-Speicher und rufst zum Schluss getOrCreate() auf, um eine laufende Session zu erhalten oder eine neue zu erstellen.

# 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

Hinweis: Falls ein FileNotFoundError auftritt wie „No such file or directory: ‘/User/YourName/Downloads/spark-2.1.0-bin-hadoop2.7/./bin/spark-submit’“, musst du deinen Spark-PATH (neu) setzen. Wechsle mit $ cd in dein Home-Verzeichnis und bearbeite die Datei .bash_profile mit $ nano .bash_profile.

Füge am Ende der Datei etwas wie Folgendes ein

export SPARK_HOME="/usr/local/spark"

Beende die Datei mit STRG + X und bestätige das Speichern mit Y. Vergiss nicht, die Änderungen mit source .bash_profile zu aktivieren.

Tipp: Du kannst bei Bedarf weitere Umgebungsvariablen setzen. Meist brauchst du sie nicht, aber gut zu wissen. Etwa so:

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

Hinweis: Du hast nun eine Standard-SparkSession initialisiert. In den meisten Fällen wirst du sie weiter konfigurieren wollen – vor allem bei echten Big-Data-Workloads. Mehr dazu hier.

Daten laden

Dieses Tutorial nutzt den California-Housing-Datensatz. Er erschien 1997 im Paper Sparse Spatial Autoregressions von Pace, R. Kelley und Ronald Barry in den Statistics and Probability Letters. Die Forschenden bauten den Datensatz auf Basis der Volkszählung in Kalifornien von 1990.

Die Daten enthalten eine Zeile pro „Census Block Group“. Eine Block Group ist die kleinste geografische Einheit, für die das U.S. Census Bureau Stichprobendaten veröffentlicht (typischerweise 600 bis 3.000 Personen). In dieser Stichprobe umfasst eine Block Group im Schnitt 1425,5 Personen in einem geografisch kompakten Gebiet. Diese Infos findest du auf dieser Seite oder im oben erwähnten Paper hier.

Die räumlichen Daten umfassen 20.640 Beobachtungen zu Immobilienpreisen mit 9 ökonomischen Variablen:

• Longitude bezeichnet die geografische Länge je Block Group;
• Latitude bezeichnet die geografische Breite je Block Group;
• Housing median age ist das Medianalter der Personen in einer Block Group. Hinweis: Der Median ist der Wert in der Mitte der Verteilung;
• Total rooms ist die Gesamtzahl der Räume in den Häusern je Block Group;
• Total bedrooms ist die Gesamtzahl der Schlafzimmer in den Häusern je Block Group;
• Population ist die Einwohnerzahl einer Block Group;
• Households bezeichnet die Anzahl der Haushalte je Block Group;
• Median income ist das Medianeinkommen der Personen in einer Block Group; und
• Median house value ist die Zielvariable und bezeichnet den medianen Hauswert je Block Group.

Zudem wurden alle Block Groups mit Nulleinträgen in unabhängigen und abhängigen Variablen aus den Daten ausgeschlossen.

Der Median house value ist die abhängige Variable und dient in deinem ML-Modell als Zielvariable.

Du kannst die Daten hier herunterladen. Suche den Ordner houses.zip, lade ihn herunter und entpacke ihn, um an die Datenordner zu kommen.

Als Nächstes liest du die Daten mit der Methode textFile() aus dem heruntergeladenen Ordner als RDDs ein. Diese Methode erwartet eine URI zur Datei (hier der lokale Pfad) und liest sie als Sammlung von Zeilen. Zur Kontrolle lädst du nicht nur die .data-Datei, sondern auch die .domain-Datei mit dem Header. So kannst du die Reihenfolge der Variablen gegenprüfen.

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

Daten erkunden

Du hast bereits viele Informationen von der Webseite mit dem Datensatz gesammelt, aber praktische Inspektionen mit Spark in Python sind unerlässlich.

Wichtig: Da die Ausführung in Spark „lazy“ ist, wurde noch nichts ausgeführt. Deine Daten wurden noch nicht wirklich gelesen. Die Variablen rdd und header sind bisher nur Konzepte. Du musst Spark „anschubsen“. Schau dir zunächst den header mit collect() an:

header.collect()

collect() holt das gesamte RDD auf eine Maschine, und du siehst etwa Folgendes:

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

Tipp: Sei vorsichtig mit collect()! Das kann den Treiber aus dem Speicher laufen lassen. take() ist sicherer, wenn du nur einen Ausschnitt sehen willst. Generell solltest du Ergebnissets begrenzen, so wie bei SQL.

Die Reihenfolge der Variablen entspricht der oben beschriebenen, und alle Spalten sind kontinuierlich. Lass Spark mehr arbeiten und sieh dir die California-Housing-Daten an, um das zu bestätigen.

Rufe take() auf deinem RDD auf:

rdd.take(2)

Damit holst du die ersten 2 Elemente des RDD. Wie erwartet: Da du mit textFile() eingelesen hast, liegen die Zeilen als Strings vor. Die Einträge sind durch Komma getrennt, ebenso die Zeilen:

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

Das musst du aufbereiten. Du musst die Einträge nicht einzeln aufspalten, aber die Zeilen sollten separate Elemente sein. Dafür verwendest du map() mit einer Lambda-Funktion, die an jedem Komma trennt. Prüfe das Ergebnis wieder mit take():

Erinnerung: Lambda-Funktionen sind anonyme Funktionen, die zur Laufzeit erzeugt werden.

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

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

Du erhältst Folgendes:

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

Alternativ kannst du auch Folgendes nutzen:

# Inspect the first line 
rdd.first()

# Take top elements
rdd.top(2)

Wenn du Pandas oder Data Frames in R gewohnt bist, erwartest du einen Header – den gibt es hier nicht. Um dir das Leben zu erleichtern, wechselst du vom RDD zu einem DataFrame. DataFrames sind RDDs vorzuziehen, wann immer möglich. Besonders in Python sind sie performanter.

Was ist der Unterschied?

RDDs nutzt du, wenn du Low-Level-Transformationen und -Actions auf unstrukturierten Daten brauchst, also kein Schema erzwingen oder Spaltennamen verwenden willst. Performancevorteile von DataFrames für (semi-)strukturierte Daten spielen dann weniger eine Rolle. RDDs sind sinnvoll, wenn du lieber mit funktionalen Konstrukten statt domänenspezifischen Ausdrücken arbeitest.

Wir wechseln jetzt zu DataFrames, um High-Level-Ausdrücke, SQL-Abfragen zur weiteren Exploration und spaltenweisen Zugriff zu nutzen.

Los geht’s.

Schritt eins: Erzeuge eine SchemaRDD bzw. ein RDD aus Row-Objekten mit Schema. Ziel ist, wie bei einem DataFrame, Zeilen und Spalten mit Datentypen zu haben.

Verwende erneut map() mit einer Lambda-Funktion, die jedes Element einer Zeile einem Feld in einer Row zuordnet. Beispiel für die erste Zeile:

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

Die Lambda-Funktion konstruiert eine Row der SchemaRDD: Index 0 heißt „longitude“, und so weiter.

Mit dieser SchemaRDD kannst du das RDD mit toDF() in ein DataFrame umwandeln.

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

Jetzt kannst du df mit Methoden wie first(), take(), head() und show() inspizieren:

# Show the top 20 rows 
df.show()

Das sieht deutlich anders aus als das RDD zuvor:

Tipp: df.columns gibt dir die Spaltennamen zurück.

Die Daten sind sauber in Spalten organisiert, aber wie sieht es mit den Datentypen aus? Spark versucht beim Einlesen ein Schema zu erkennen – hat das geklappt? Nutze df.dtypes oder df.printSchema(), um mehr über die Datentypen zu erfahren.

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

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

Da du die erste Zeile nicht ausführst, bekommst du Folgendes zurück:

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 Spalten sind noch vom Typ String … Ernüchternd!

Wenn du mit diesem DataFrame weiterarbeiten willst, solltest du passendere Datentypen vergeben. Das verbessert auch die Performance. Intuitiv könntest du alle Spalten von df auf FloatType() casten:

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

Diese Wiederholungen sind jedoch fehleranfällig und nicht schön zu lesen. Warum nicht eine Funktion schreiben, die das sauber für dich erledigt?

Die folgende Funktion nimmt ein DataFrame, Spaltennamen und den neuen Datentyp. Für jeden Spaltennamen castest du die Spalte auf den neuen Typ und gibst das DataFrame zurück:

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

So wirkt es deutlich aufgeräumter! Prüfe die Datentypen von df erneut mit printSchema().

Jetzt geht es an die eigentliche Exploration. Spaltenweiser Zugriff und SQL-Abfragen sind zwei große Vorteile von DataFrames. Starte klein und wähle zwei Spalten aus df, von denen du 10 Zeilen sehen möchtest:

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

Das liefert dir:

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

Du kannst die Abfragen natürlich komplexer gestalten, z. B. so:

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

Ergebnis:

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

Neben Abfragen kannst du auch Zusammenfassungsstatistiken ausgeben. Das hilft dir im nächsten Schritt:

df.describe().show()

Sieh dir Minimum und Maximum der numerischen Attribute an. Mehrere Spalten haben große Wertebereiche – du solltest deinen Datensatz normalisieren.

Datenvorverarbeitung

Auf Basis deiner kurzen explorativen Analyse weißt du genug, um die Daten für das Modell aufzubereiten.

• Um fehlende Werte musst du dich nicht kümmern; alle Nullwerte wurden ausgeschlossen.
• Du solltest die Daten standardisieren, da die Spannweiten groß sind.
• Zusätzliche Merkmale sind sinnvoll, z. B. Schlafzimmer pro Raum oder Räume pro Haushalt.
• Auch die Zielvariable ist recht groß; zur einfacheren Handhabung passt du ihre Werte etwas an.

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