Einführung in Spark – ein Text Mining-Projekt

Grundsätzlich baut Spark auf einer Master/Slave-Architektur auf, in der ein Master-Knoten (genannt Clustermanager) wirkt und Aufgaben auf die Slaves (genannt Worker-Knoten) verteilt. Der Spark-Driver ist ein Prozess, der den Spark Context bereitstellt und unser Zugang zur gesamten Spark-Engine. In unserem Beispiel läuft der Spark-Driver auf dem Master-Knoten (dies muss er nicht zwingend). Dazu benötigt Spark noch einen Clustermanager und ein verteiltes Speichersystem. Als Clustermanager stehen Hadoop YARN oder Mesos zur Auswahl, als Speichersystem HDFS, MapR-FS oder viele weitere. Zum Auspropieren und für Entwicklungszwecke gibt es Spark aber auch in einem lokalen Stand Alone-Modus, in dem Clustermanager und spezielles Dateisystem wegfallen können.

Spark stellt drei grundlegende Datenstrukturen zur Verfügung: den resilient distributed dataset (RDD), den DataFrame (DF), und den DataSet (DS).

RDDs sind die älteste der Datenstrukturen. Wie der Name schon sagt, ist ein RDD eine über viele Knoten verteilte Datensammlung, man kann sie sich wie eine Collection von Records (streng typisiert) vorstellen, die verteilt über den Cluster gespeichert wird. Dadurch können Operationen parallelisiert und fehlertolerant (resilient) ausgeführt werden.

In neueren Spark-Versionen stehen zusätzlich DataFrames und DataSets zur Verfügung – beide kann man sich wie eine SQL-Tabelle vorstellen, bestehend aus Spalten, die Namen für den einfachen Zugriff darauf haben. Der Vorteil von DataSets gegenüber DataFrames ist, dass diese über Compilezeit-Typsicherheit verfügen. Diesen Vorteil können sie natürlich nur bei statisch typisierten Sprachen wie Java oder Scala ausspielen, für Python (das wir in diesem Artikel verwenden wollen), sind DataFrames die passende Abstraktion.

Alle drei Datenstrukturen sind unveränderlich (immutable). Neue Objekte werden entweder direkt aus einer Datenquelle wie einem Textfile eingelesen oder aus einer bestehenden Nicht-Spark-Datenstruktur (wie einer Liste oder einem Set) erzeugt.

Spark stellt vielfältige Operationen auf den RDDs/DFs/DSs zur Verfügung. Dabei ist zwischen Transformationen und Aktionen zu unterscheiden:

• Transformationen erzeugen aus einem RDD/DF/DS einen neuen RDD/DF/DS. Die Transformation filter() zum Beispiel erzeugt einen neuen RDD/DF/DS, der nur die Zeilen enthält, die eine angegebene Bedingung erfüllen; withColumn() wird verwendet, um eine neue Spalte zu einem DF/DS anzufügen, typischerweise abgeleitet aus den Werten einer oder mehrerer bestehender Spalten (d. h. es wird ein neuer DF/DS mit einer zusätzlichen Spalte zurückgegeben). Auch Joins/Unions/Projektionen und viele weitere Operationen sind möglich.
• Aktionen auf der anderen Seite berechnen einen "Wert", d. h. sie geben keinen RDD/DF/DS zurück. Beispiele dafür sind first() und head(), um ein oder einige wenige Elemente auszulesen, show() für einen tabellenformatierten Überblick eines DF oder count(), um die Anzahl Zeilen des RDD/DF/DS zu berechnen.

Der entscheidende Unterschied zwischen Transformationen und Aktionen ist, dass bei Transformationen die "lazy evaluation" angewendet werden, während Aktionen direkt berechnet werden: das Anwenden einer Transformation benötigt so gut wie keine Zeit, Spark merkt sich lediglich, wie der neue RDD/DF/DS aus dem Bestehenden berechnet werden soll – und das über beliebig viele Transformationen hinweg. Dadurch entsteht die sogenannte "lineage", also die Entstehungsgeschichte eines RDD/DF/DS. Erst das Ausführen einer Aktion bewirkt, dass diese Entstehungsgeschichte nun wirklich (verteilt über den Spark-Cluster) abgearbeitet wird.

Beim Abarbeiten der Entstehungsgeschichte/Lineage zeigt sich dann ein weiterer Unterschied zwischen RDD und DF/DS: bei DF/DS kommt ein Query-Optimizer (ähnlich dem einer relationalen Datenbank) zum Einsatz, bei RDDs nicht. Dadurch sind DF/DS den RDD häufig aus Performancegesichtspunkten deutlich überlegen.

Wir werden im Folgenden hauptsächlich mit DataFrames arbeiten (ein Umwandeln zwischen den Datenstrukturen ist jedoch jederzeit einfach möglich). Einige der Eigenschaften und Unterschiede der drei Datenstrukturen sind in den folgenden Tabellen zusammengefasst.

Eine einfache Option, aus Patenttexten Vektoren zu erzeugen, ist, die in allen Patenten zusammen vorkommenden Terme in beliebiger Reihenfolge anzuordnen, für jedes Patent zu zählen, wie oft jeder Term darin vorkommt, und diese Zahl als Wert der entsprechenden Dimension zu verwenden. Dies ist ein guter erster Schritt, da die Höhe des Wertes eines Terms (die term frequency, TF) darüber Aufschluss gibt, wie häufig der Term im Patent vorkommt, also vermutlich wie wichtig er für den Inhalt dieses Patent ist (vgl. Abb. 2: Beispielpatent als Liste von Termen und Abb. 3: TF (nicht normalisiert)). Allerdings sind wir damit noch nicht fertig.

Ein weiterer Aspekt, den wir berücksichtigen wollen, ist, dass Patente verschieden lang sind. Die TFs (Anzahl Vorkommen eines Terms) sind dadurch für ein langes Patent höher als für ein kurzes Patent, selbst wenn genau dieselben Terme in gleicher relativer Häufigkeit in den beiden Patenten vorkommen. Dies ist unschön, aber sehr leicht zu lösen: wir normieren den Vektor jedes Patents (z. B. so, dass die darin enthaltenen TFs in Summe 1 ergeben). Damit erhalten wir die normierte term frequency (vgl. Abb. 4: TF (normalisiert))

Weiterhin beobachten wir, dass manche Terme in sehr vielen Patenten vorkommen (z. B. wird der Term "claim" in fast jedem Patent vorkommen, da er nicht in unserer Stoppwortliste enthalten ist), und uns somit nur wenig helfen, die Patente zu unterscheiden, während andere Terme in nur wenigen Patenten vorkommen, also vermutlich ein starkes Anzeichen für die Ähnlichkeit dieser Patente und den Unterschied zu allen anderen Patenten sind. Wir definieren die IDF oder inverse document frequency eines Terms als log(N/n), wobei N die Anzahl aller Patente ist, n die Anzahl von Patenten in denen der Term vorkommt. Terme die in fast allen Patenten vorkommen, haben eine IDF von nahezu 0. In umso weniger Patenten ein Term vorkommt, desdo höher ist sein IDF-Wert (vgl. Abb. 5: IDF-Werte).

Den finalen TF-IDF-Vektor für jedes Patent erzeugen wir, indem wir jeden (normierten) TF-Wert des Patentes mit dem IDF-Wert des Terms multiplizieren, d. h. den Term mit seiner "Unterscheidungskraft" bewerten (vgl. Abb. 6: TF-IDF-Matrix).

Alle Berechnungen können wir leicht mit den Funktionen durchführen, die uns MLlib bereitstellt. Zunächst zählen wir, welche Terme wie oft in jedem Patent vorkommen und speichern diese Information als Vektor für das Patent. Diese counts teilen wir durch die Anzahl der Terme des Patents, normalisieren also diesen Vektor. Die IDF berechnen wir aus den term frequencies. Dazu wird gezählt, in wie vielen der Dokumentvektoren der count eines Terms ungleich 0 ist und dieser Term entsprechend gewichtet. 

CountVectorizer() muss hierbei zweimal über den DataFrame iterieren. Einmal, um herauszufinden, welche Worte insgesamt vorkommen und einen entsprechend dimensionierten Vektor bereitzustellen, was mit der Funktion fit() geschieht, und ein zweites Mal, um per transform() die Gewichte zu setzen. Genauso muss IDF() zuerst die IDF-Gewichte jedes Wortes per fit() finden, um sie dann per transform() auf die Dokumentvektoren anzuwenden, also mit der TF zu multiplizieren.

from pyspark.ml.feature import CountVectorizer
from pyspark.ml.feature import Normalizer
from pyspark.ml.feature import IDF

cv = CountVectorizer(inputCol="claimStemmed", outputCol="claimTF", vocabSize=vSize)
cvModel = cv.fit(df)
tf_df = cvModel.transform(df)

normalizer = Normalizer(inputCol="claimTF", outputCol="claimTFN")
tfn_df = normalizer.transform(tf_df)

idf = IDF(inputCol="claimTFN", outputCol="IDF")

idfModel = idf.fit(tfn_df)
tfidf_df = idfModel.transform(tfn_df)

Die Tokens mit den höchsten TF-IDF-Werten identifizieren in der Regel aussagekräftige Schlagworte. Beispiele für einige zufällig gewählte Patente und die besten Tokens, die aus den Patentbeschreibungen extrahiert wurden, sehen wir hier:

Beide Probleme können wir jedoch elegant lösen, indem wir an Stelle der Euklidischen Distanz die Kosinus-Ähnlichkeit verwenden. Wenn wir uns die TF-IDF-Vektoren als "Pfeile" vom Ursprung aus vorstellen, beschreibt die Richtung des Pfeils das inhaltliche Thema des Patents. Der Winkel zwischen zwei solchen "Patent-Pfeilen" beschreibt, wie ähnlich sich die beiden Patente sind – ist der Winkel 0 Grad oder nahe an 0, gehen die Patente in die gleiche Richtung, ist er 90 Grad oder nahe daran, sind die Patent so unterschiedlich wie möglich. Der Winkel liegt also zwischen 0 und (absolut) 90 Grad (vgl. Abb. 7: Patente und Winkel).

Wenn wir auf diesen Winkel noch die Kosinusfunktion anwenden, erhalten wir eine Zahl zwischen 1 = cos(0 Grad) bei maximaler Ähnlichkeit und 0 = cos(90 Grad) bei maximaler Unähnlichkeit (vgl. Abb. 8: Kosinusfunktion). Dies ist ein sehr schönes Ähnlichkeitsmaß, und lässt sich noch dazu sehr einfach berechnen (zur Erinnerung aus der Mathematik: der Kosinus zweier Vektoren eines Euklidischen Vektorraums ist das Skalarprodukt der beiden geteilt durch das Produkt der Normen). 

Wir benötigen zwei Funktionen: Eine user-defined function zur Berechnung der Kosinus-Ähnlichkeit zweier Vektoren und eine Funktion, welche die Ähnlichkeit zwischen einem Suchtext und allen existierenden Patenten zurückgibt. Als Suchtext wählen wir dabei eines der Patente in unserem Datensatz aus.

from pyspark.sql.types import FloatType

# Berechne die cosine-simliarity: dot product / product der L2-Norms
def csim(vec1, vec2):
    return vec1.dot(vec2)/vec1.norm(2)*vec2.norm(2))

# Berechne die Ähnlichkeiten zu einem gegebenen Patent
def getSimilarities(df, id):
    patent = df.where("id == "+id)
    my_tfidf = patent.select("IDF").first().IDF
    csim_udf = udf(lambda x: round(float(csim(my_tfidf, x)),2))
    sim_df = df.withColumn("similarity", csim_udf("IDF"))
    return sim_df

Wir können natürlich nicht per Hand alle Suchergebnisse überprüfen. Aber zumindest ein Überblick über die Verteilung aller Distanzen hilft uns, die Suchergebnisse zu validieren. Dazu berechnen wir die Ähnlichkeiten und transformieren den DataFrame in einen RDD, der nur noch die Zahlenwerte enthält. Dieses lassen wir uns mit collect() als Python-Liste zurückgeben. Zum Plotten eines Histogramms verwenden wir matplotlib. Wir erkennen einige wenige Ergebnisse mit hoher Ähnlichkeit, von denen wir schon gesehen haben, dass sie recht gut zu unserem Suchtext passen. Der weitaus größte Teil hat offensichtlich eine geringe Ähnlichkeit zu "unserem" Patent.

import matplotlib.pyplot as plt

similarities = getMostSimilar(tfidf_df, "10")
.select(["similarity"])
.rdd.map(lambda row: float(row['similarity'])).collect()

plt.hist(similarities)
plt.yscale('log', nonposy='clip')
plt.title("Verteilung der Ähnlichkeitswerte für ein Patent"
plt.xlabel("Kosinus-Ähnlichkeit")
plt.ylabel("Häufigkeit")

plt.show()