Auf einen Blick: Lerne, User Defined Functions (UDFs) in Python für Apache Spark zu schreiben, um die Datenverarbeitung zu verbessern. Das Tutorial erklärt drei Methoden: Verwendung von withColumn(), SQL-Anweisungen und RDD-Map-Transformationen.
- UDFs ermöglichen es, benutzerdefinierte Funktionen zu registrieren und auf Spark-Datenframes anzuwenden.
- Codebeispiele veranschaulichen die Implementierung jeder Methode effektiv.
In diesem Artikel werde ich erklären, wie man benutzerdefinierte Funktionen (UDF) in Python für Apache Spark schreibt. Der Code für dieses Beispiel ist hier.
Warum brauchst du UDFs?
Spark speichert Daten in Dataframes oder RDDs – resiliente, verteilte Datensätze. Man kann sich diese wie Datenbanken vorstellen. Wie bei einer traditionellen SQL-Datenbank, z. B. mySQL, kann man keine eigene benutzerdefinierte Funktion erstellen und diese direkt gegen die Datenbank ausführen. Man muss die Funktion zuerst registrieren . Das heißt, sie in der Datenbank speichern, als wäre sie eine der integrierten Datenbankfunktionen, wie sum(), average, count() usw.
Das ist bei Spark-Dataframes der Fall. Mit Spark-RDDs kann man Funktionen direkt an den Zeilen eines RDDs ausführen.
Drei Ansätze für UDFs
Es gibt drei Möglichkeiten, UDFs zu erstellen:
- df = df.withColumn
- DF = sqlContext.sql("SQL-Anweisung aus <DF>")
- rdd.map(customFunction())
Wir zeigen unten die drei Ansätze, beginnend mit dem ersten.
Ansatz 1: withColumn()
Im Folgenden erstellen wir einen einfachen Dataframe und RDD. Wir schreiben eine Funktion, um das einzige Textfeld in der Datenstruktur in eine ganze Zahl umzuwandeln. Das ist etwas, das Sie tun könnten, wenn Sie zum Beispiel mit maschinellem Lernen arbeiten, bei dem alle Daten in Zahlen umgewandelt werden müssen, bevor Sie sie in einen Algorithmus einbinden.
Beachten Sie die untenstehenden Importe. Beziehen Sie sich auf diese in jedem Beispiel, damit Sie wissen, welches Objekt Sie für jeden der drei Ansätze importieren müssen.
Nachfolgend finden Sie den vollständigen Code für Ansatz 1. Zuerst betrachten wir die wichtigsten Abschnitte. Erstellen Sie einen Dataframe mit dem üblichen Ansatz:
df = spark.createDataFrame(data,schema=schema)
Jetzt machen wir zwei Dinge. Erstens erstellen wir eine Funktion colsInt und registrieren sie. Diese registrierte Funktion ruft eine weitere Funktion toInt() auf, die wir nicht registrieren müssen. Das erste Argument in udf.register("colsInt", colsInt) ist der Name, den wir für die Funktion verwenden. Das zweite ist die Funktion, die wir registrieren wollen.
colsInt = udf(lambda z: toInt(z), IntegerType())
spark.udf.register("colsInt", colsInt)
def toInt(s):
if isinstance(s, str) == True:
st = [str(ord(i)) for i in s]
return(int(''.join(st)))
else:
return Null
Dann nennen wir die Funktion colinsInt, so. Das erste Argument ist der Name der neuen Spalte, die wir erstellen wollen. Das zweite ist die Spalte im Dataframe, die in die Funktion eingefügt werden soll.
df2 = df.withColumn( 'semployee',colsInt('employee'))
Denken Sie daran, dass df['employees'] ein Spaltenobjekt ist, kein einzelner Mitarbeiter. Das bedeutet, wir müssen alle Zeilen dieser Spalte in eine Schleife laufen lassen – also verwenden wir diese Lambda (Inline-)Schleife.
colsInt = udf(lambda z: toInt(z), IntegerType())
Hier ist Ansatz 1 insgesamt:
import pyspark
from pyspark import SQLContext
from pyspark.sql.types import StructType, StructField, IntegerType, FloatType, StringType
from pyspark.sql.functions import udf
from pyspark.sql import Row
conf = pyspark.SparkConf()
sc = pyspark.SparkContext.getOrCreate(conf=conf)
spark = SQLContext(sc)
schema = StructType([
StructField("sales", FloatType(),True),
StructField("employee", StringType(),True),
StructField("ID", IntegerType(),True)
])
data = [[ 10.2, "Fred",123]]
df = spark.createDataFrame(data,schema=schema)
colsInt = udf(lambda z: toInt(z), IntegerType())
spark.udf.register("colsInt", colsInt)
def toInt(s):
if isinstance(s, str) == True:
st = [str(ord(i)) for i in s]
return(int(''.join(st)))
else:
return Null
df2 = df.withColumn( 'semployee',colsInt('employee'))
Jetzt zeigen wir die Ergebnisse an. Beachten Sie, dass der neue Spalten-Mitarbeiter hinzugefügt wurde. withColumn() erstellt einen neuen Dataframe, daher haben wir df2 erstellt.
df2.show()
+-----+--------+---+----------+
|sales|employee| ID| semployee|
+-----+--------+---+----------+
| 10.2| Fred|123|1394624364|
+-----+--------+---+----------+
Ansatz 2: Verwendung von SQL
Der erste Schritt ist, den Dataframe als Tabelle zu registrieren, damit wir SQL-Anweisungen darauf ausführen können. DF ist der Dataframe und DFTAB ist die temporäre Tabelle, die wir erstellen.
spark.registerDataFrameAsTable(df, "dftab")
Jetzt erstellen wir einen neuen Dataframe df3 aus dem bestehenden auf df und wenden die colsInt-Funktion auf die Mitarbeiterspalte an.
df3 = spark.sql("select sales, employee, ID, colsInt(employee) as iemployee from dftab")
Hier sind die Ergebnisse:
df3.show()
+-----+--------+---+----------+
|sales|employee| ID| iemployee|
+-----+--------+---+----------+
| 10.2| Fred|123|1394624364|
+-----+--------+---+----------+
Ansatz 3: RDD-Karte
Ein Dataframe besitzt keine map()-Funktion. Wenn wir diese Funktion verwenden wollen, müssen wir den Dataframe mit dff.rd in ein RDD umwandeln.
Wenden Sie die Funktion folgendermaßen an:
rdd = df.rdd.map(toIntEmployee)
Dies übergibt ein Zeilenobjekt an die Funktion toIntEmployee. Wir müssen also ein Zeilenobjekt zurückgeben. Das RDD ist unveränderlich, daher müssen wir eine neue Zeile erstellen.
Im Folgenden beziehen wir uns auf das Mitarbeiterelement in der Zeile beim Namen und wandeln dann jeden Buchstaben in dieses Feld in eine ganze Zahl um und fügen diese aneinander.
def toIntEmployee(rdd):
s = rdd["employee"]
if isinstance(s, str) == True:
st = [str(ord(i)) for i in s]
e = int(''.join(st))
else:
e = s
return Row(rdd["sales"],rdd["employee"],rdd["ID"],e)
Jetzt drucken wir die Ergebnisse aus:
for x in rdd.collect():
print(x)
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.