Auf einen Blick: Die Verwendung von Python mit Spark ML erleichtert die Entwicklung von Klassifikationsmodellen, insbesondere zur Vorhersage von Gesundheitszuständen wie Herzfehlern. Das Tutorial beschreibt Methoden zur Datenvorbereitung, -transformation und zur Implementierung der logistischen Regression.
- Nutzt Sparks DataFrame für die effiziente Handhabung großer Datensätze mit Python-Bibliotheken.
- Geht durch den gesamten Workflow und erreicht eine Modellgenauigkeit von 93 %.
Wir haben darüber geschrieben, wie man Spark ML mit der Scala-Programmiersprache nutzt. Aber nicht viele Programmierer kennen Scala. Python hat Java in Bezug auf die Anzahl der Nutzer überholt, was vor allem auf der Stärke des maschinellen Lernens basiert. Also richten wir unsere Aufmerksamkeit auf die Nutzung von Spark ML mit Python.
Man könnte sagen, dass Spark Scala-zentriert ist. Scala hat sowohl Python- als auch Scala-Schnittstellen sowie Kommandozeilen-Interpreter. Scala ist die Standard-Variante. Die Python-Variante heißt pyspark. Die meisten Beispiele, die Spark gibt, sind in Scala und in manchen Fällen keine Beispiele in Python.
Apache-Atom
Python ist die bevorzugte Sprache für Data Science wegen NumPy, Pandas und matplotlib, die Werkzeuge sind, die das Arbeiten mit Arrays und Diagrammzeichnungen erleichtern und mit großen Datenarrays effizient arbeiten können. Aber Spark ist darauf ausgelegt, mit enormen Datenmengen zu arbeiten, die über einen Cluster verteilt sind. Es ist gute Praxis, beide Tools zu nutzen und je nach Nachfrage hin und her zu wechseln.
Aber wie wir sehen werden, da Spark Dataframe nicht dasselbe ist wie ein Pandas-Dataframe, besteht zwischen all diesen Objekten keine 100%ige Kompatibilität. Man muss Spark-Dataframes in Listen, Arrays und andere Strukturen umwandeln, um sie mit Matplotlib zu plotten. Da man Arrays nicht mit dem vertrauten [:,4] slicen kann, benötigt man mehr Code, um denselben Vorgang auszuführen.
Aber das andere Problem ist die Performance. Apache Atom existiert, um Objekte in Java-Prozessen effizient in Python-Prozesse und umgekehrt umzuwandeln. Spark ist in Java und Scala geschrieben. Scala basiert auf Java. Python läuft natürlich in einem Python-Prozess.
Arrow beschleunigt die Operationen durch die Umwandlung von Spark-Datenframes in Pandas-Datenrahmen und mit spaltenweisen Operationen wie . withcolumn().
Spark erläutert einige der damit verbundenen Probleme und die Konfigurationsänderungen, die man in Spark vornehmen muss, um von diesem Leistungsschub zu profitieren, in ihrer Apache Arrow-Dokumentation .
Herzpatientendaten
Laden Sie die Daten aus dem Datensatz der Universität São Paolo herunter, der hier verfügbar ist. Wenn Sie neugierig sind, sehen Sie sich diese Diskussion an.
Die Säulen sind:
- Alter
- Geschlecht
- Brustschmerztyp (4 Werte)
- Ruheblutdruck
- Serumcholesterin in mg/dl
- Nüchternblutzucker > 120 mg/dl
- Ruhe-elektrokardiographische Ergebnisse (Werte 0,1,2)
- Maximale erreichte Herzfrequenz
- Bewegungsbedingte Angina
- Oldpeak = ST-Depression, ausgelöst durch Bewegung im Vergleich zur Ruhe
- Steigung des Peak-Übungs-ST-Segments
- Anzahl der Hauptgefäße (0–3), die durch Fluoroskopie gefärbt sind
- Thal: 3 = normal; 6 = behobener Defekt; 7 = reversierbarer Defekt
Der Code, erklärt
Das Ziel ist es, ein prädiktives binäres logistisches Regressionsmodell mit Spark ML und Python zu erstellen, das vorhersagt, ob jemand einen Herzfehler hat. Der folgende Code ist hier in einem Zeppelin-Notebook verfügbar.
Zuerst lesen wir die Daten ein und weisen Spaltennamen zu. Da die Daten klein sind und Pandas einfacher ist, lesen wir sie in einen Pandas-Datenrahmen ein. Dann konvertieren wir ihn mit spark.createDataFrame() in einen Spark-Datenrahmen.
Du verstehst vielleicht, was ich meine, dass der Spark-Dataframe einige der Funktionen von Pandas fehlt. Insbesondere verwenden wir Pandas, damit wir . iloc() verwenden können, um die ersten 13 Spalten zu nehmen und die letzte wegzulassen, was offenbar Rauschen ist, das nicht für die Daten gedacht ist.
%spark.pyspark
import pandas as pd
from pyspark.sql.types import StructType, StructField, NumericType
cols = ('age',
'sex',
'chest pain',
'resting blood pressure',
'serum cholesterol',
'fasting blood sugar',
'resting electrocardiographic results',
'maximum heart rate achieved',
'exercise induced angina',
'ST depression induced by exercise relative to rest',
'the slope of the peak exercise ST segment',
'number of major vessels ',
'thal',
'last')
data = pd.read_csv('/home/ubuntu/Downloads/heart.csv', delimiter=' ', names=cols)
data = data.iloc[:,0:13]
data['isSick'] = data['thal'].apply(isSick)
df = spark.createDataFrame(data)
Das Feldthal zeigt an, ob der Patient ein Herzproblem hat. Die Zahlen sind wie folgt:
- Ein Wert von 3 bedeutet, dass der Patient gesund (normal) ist.
- Ein Wert von 6 bedeutet, dass das Gesundheitsproblem des Patienten behoben wurde.
- Ein Wert von 7 bedeutet, dass das Gesundheitsproblem des Patienten behoben werden kann.
Schreiben Sie also diese Funktion isSick() so, dass 0 als negativ und 1 als positiv gekennzeichnet wird, da binäre logistische Regression eines von zwei Ergebnissen erfordert.
def isSick(x):
if x in (3,7):
return 0
else:
return 1
Mit maschinellem Lernen und Klassifikations- oder Regressionsproblemen haben wir:
- Eine Matrix von Merkmalen, einschließlich Alter des Patienten, Blutzucker usw.
- Ein Vektor von Labels, der angibt, ob der Patient ein Herzproblem hat.
Da wir ein Zeppelin-Notizbuch verwenden und PySpark die Python-Befehlsshell für Spark ist, schreiben wir %spark.pyspark oben in jeder Zeppelin-Zelle, um die Sprache und den Interpreter anzugeben, den wir verwenden möchten.
Als Nächstes geben wir an, welche Spalten im df-Dataframe wir als Features verwenden möchten. Dann verwenden wir den VectorAasembler , um alle zwölf dieser Felder in eine neue Spalte namens Features zu packen, die alle diese als Array enthält.
Nun erstellen wir den Spark-Datenrahmen raw_data verwenden wir die Operation transform() und wählen nur die Spalte Features aus.
%spark.pyspark
from pyspark.ml.feature import StandardScaler
from pyspark.ml.feature import VectorAssembler
features = ('age',
'sex',
'chest pain',
'resting blood pressure',
'serum cholestoral',
'fasting blood sugar',
'resting electrocardiographic results',
'maximum heart rate achieved',
'exercise induced angina',
'ST depression induced by exercise relative to rest',
'the slope of the peak exercise ST segment',
'number of major vessels ')
assembler = VectorAssembler(inputCols=features,outputCol="features")
raw_data=assembler.transform(df)
raw_data.select("features").show(truncate=False)
Wir verwenden den Standard Scaler, um alle Zahlen auf derselben Skala zu setzen, was Standardpraxis im maschinellen Lernen ist. Dieser nimmt die Beobachtung, subtrahiert den Mittelwert und teilt diesen dann durch die Standardabweichung.
%spark.pyspark
from pyspark.ml.feature import StandardScaler
standardscaler=StandardScaler().setInputCol("features").setOutputCol("Scaled_features")
raw_data=standardscaler.fit(raw_data).transform(raw_data)
raw_data.select("features","Scaled_features").show(5)
So sehen die Feature-Daten jetzt aus:
+--------------------------------------------------------+
|features |
+--------------------------------------------------------+
|[70.0,1.0,4.0,130.0,322.0,0.0,2.0,109.0,0.0,2.4,2.0,3.0]|
|[67.0,0.0,3.0,115.0,564.0,0.0,2.0,160.0,0.0,1.6,2.0,0.0]|
|[57.0,1.0,2.0,124.0,261.0,0.0,0.0,141.0,0.0,0.3,1.0,0.0]|
|[64.0,1.0,4.0,128.0,263.0,0.0,0.0,105.0,1.0,0.2,2.0,1.0]|
|[74.0,0.0,2.0,120.0,269.0,0.0,2.0,121.0,1.0,0.2,1.0,1.0]|
|[65.0,1.0,4.0,120.0,177.0,0.0,0.0,140.0,0.0,0.4,1.0,0.0]|
|[56.0,1.0,3.0,130.0,256.0,1.0,2.0,142.0,1.0,0.6,2.0,1.0]|
|[59.0,1.0,4.0,110.0,239.0,0.0,2.0,142.0,1.0,1.2,2.0,1.0]|
|[60.0,1.0,4.0,140.0,293.0,0.0,2.0,170.0,0.0,1.2,2.0,2.0]|
|[63.0,0.0,4.0,150.0,407.0,0.0,2.0,154.0,0.0,4.0,2.0,3.0]|
|[59.0,1.0,4.0,135.0,234.0,0.0,0.0,161.0,0.0,0.5,2.0,0.0]|
|[53.0,1.0,4.0,142.0,226.0,0.0,2.0,111.0,1.0,0.0,1.0,0.0]|
Wie üblich teilen wir die Daten in Trainings - und Testdatensätze auf. Wir haben nicht viele Daten, daher verwenden wir eine 50/50-Aufteilung.
%spark.pyspark
from pyspark.ml.tuning import ParamGridBuilder, TrainValidationSplit
training, test = raw_data.randomSplit([0.5, 0.5], seed=12345)
Jetzt erstellen wir das logistische Regressionsmodell und trainieren es, das heißt, das Modell berechnet die Koeffizienten und den Intercept, die am nächsten den Ergebnissen in der Label-Spalte isSick entsprechen
%spark.pyspark
from pyspark.ml.classification import LogisticRegression
lr = LogisticRegression(labelCol="isSick", featuresCol="Scaled_features",maxIter=10)
model=lr.fit(training)
predict_train=model.transform(training)
predict_test=model.transform(test)
predict_test.select("isSick","prediction").show(10)
Hier zeigen wir die ersten paar Reihen im nebeneinanderfolgenden Vergleich. Diese sind größtenteils korrekt.
+------+----------+
|isSick|prediction|
+------+----------+
| 0| 0.0|
| 1| 0.0|
| 0| 0.0|
| 0| 0.0|
| 0| 0.0|
| 0| 0.0|
| 0| 0.0|
| 0| 1.0|
| 0| 1.0|
| 0| 0.0|
+------+----------+
Dies zeigt die Koeffizienten und den Intercept.
Hier sind sie:
%spark.pyspark
print("Multinomial coefficients: " + str(model.coefficientMatrix))
print("Multinomial intercepts: " + str(model.interceptVector))
Hier sind sie:
Multinomial coefficients: DenseMatrix([[-0.41550466, 1.21573123, 0.16600826, 0.36478609, 0.33716549,
-0.020686 , -0.2092542 , -0.86514924, 0.1427418 , -0.3610241 ,
0.57324392, 0.42563706]])
Multinomial intercepts: [-0.2767309778166021]
Jetzt verwenden wir einige Spark SQL-Funktionen F, um eine neue Spalte zu erstellen, die korrekt ist, wenn IsSick der Vorhersage entspricht, was bedeutet, dass das vorhergesagte Ergebnis den tatsächlichen Ergebnissen entspricht.
%spark.pyspark
import pyspark.sql.functions as F
check = predict_test.withColumn('correct', F.when(F.col('isSick') == F.col('prediction'), 1).otherwise(0))
check.groupby("correct").count().show()
Hier sind die Ergebnisse:
+-------+-----+
|correct|count|
+-------+-----+
| 1| 137|
| 0| 10|
+-------+-----+
Die Genauigkeit beträgt also 137 / 137 + 10 = 93 %
Es gibt andere Möglichkeiten, die Genauigkeit des Modells zu zeigen, wie zum Beispiel die Fläche unter der Kurve. Aber das ist am einfachsten zu verstehen, es sei denn, Sie sind erfahrener Data Scientist und Statistiker. Wir werden in zukünftigen Artikeln komplexere Wege zur Überprüfung der Genauigkeit erklären.
Apache Spark Leitfaden
- Apache Sparks maschinelle Lernpipeline: Eine Einführung
- SGD-Beispiel für lineare Regression mit Apache Spark
- Spark Decision Tree Klassifikator
- Mit logistischer Regression, Scala und Spark
- Wie man Jupyter-Notizbücher mit Apache Spark verwendet
- Verwendung von Python und Spark Machine Learning zur Klassifikation
- Wie man Spark UDFs (benutzerdefinierte Funktionen) in Python schreibt
- Python Spark ML K-Means Beispiel
- Wie man maschinelles Lernen auf die Cybersicherheit anwendet
- Apache Spark: Zusammenarbeit mit Streams
- K-bedeutet Clustering mit Apache Spark
- Spark mit Hive verwenden
- Wie man Apache Spark verwendet, um Vorhersagen für die vorbeugende Wartung zu treffen.
Diese Beiträge sind meine eigenen und spiegeln nicht unbedingt die Position, Strategien oder Meinung von BMC wider.