(Warum) müssen wir den Cache aufrufen oder auf einer RDD bestehen bleiben?

Wenn ein ausfallsicheres verteiltes Dataset (RDD) aus einer Textdatei oder -sammlung (oder aus einem anderen RDD) erstellt wird, müssen wir "Cache" oder "Persist" explizit aufrufen, um die RDD-Daten im Speicher zu speichern? Oder werden die RDD-Daten standardmäßig verteilt im Speicher gespeichert?

val textFile = sc.textFile("/user/emp.txt")

Nach meinem Verständnis ist textFile nach dem obigen Schritt eine RDD und im gesamten / einigen Speicher des Knotens verfügbar.

Wenn ja, warum müssen wir dann "cache" oder "persist" auf textFile RDD aufrufen?

Die meisten RDD-Operationen sind faul. Stellen Sie sich eine RDD als Beschreibung einer Reihe von Operationen vor. Ein RDD sind keine Daten. Also diese Zeile:

val textFile = sc.textFile("/user/emp.txt")

Es macht nichts. Es wird eine RDD erstellt, die besagt, dass diese Datei geladen werden muss. Die Datei wird zu diesem Zeitpunkt nicht geladen.

RDD-Vorgänge, bei denen der Inhalt der Daten überwacht werden muss, können nicht verzögert werden. (Diese werden als Aktionen bezeichnet .) Ein Beispiel ist RDD.count: Um die Anzahl der Zeilen in der Datei anzugeben, muss die Datei gelesen werden. Wenn Sie also schreiben textFile.count, wird zu diesem Zeitpunkt die Datei gelesen, die Zeilen werden gezählt und die Anzahl wird zurückgegeben.

Was ist, wenn Sie textFile.counterneut anrufen ? Das Gleiche: Die Datei wird erneut gelesen und gezählt. Es wird nichts gespeichert. Ein RDD sind keine Daten.

Was macht RDD.cachealso? Wenn Sie textFile.cacheden obigen Code hinzufügen :

val textFile = sc.textFile("/user/emp.txt")
textFile.cache

Es macht nichts. RDD.cacheist auch eine faule Operation. Die Datei wird immer noch nicht gelesen. Aber jetzt sagt das RDD "Lies diese Datei und speichere dann den Inhalt zwischen". Wenn Sie dann textFile.countdas erste Mal ausführen , wird die Datei geladen, zwischengespeichert und gezählt. Wenn Sie textFile.countein zweites Mal aufrufen , verwendet der Vorgang den Cache. Es werden nur die Daten aus dem Cache genommen und die Zeilen gezählt.

Das Cache-Verhalten hängt vom verfügbaren Speicher ab. Wenn die Datei beispielsweise nicht in den Speicher passt, textFile.countwird auf das übliche Verhalten zurückgegriffen und die Datei erneut gelesen.

Ich denke, die Frage wäre besser formuliert als:

Wann müssen wir den Cache aufrufen oder auf einer RDD bestehen bleiben?

Spark-Prozesse sind faul, das heißt, nichts wird passieren, bis es erforderlich ist. Um die Frage schnell zu beantworten val textFile = sc.textFile("/user/emp.txt"), passiert nach der Ausgabe nichts mit den Daten, sondern nur eine HadoopRDD, die die Datei als Quelle verwendet.

Nehmen wir an, wir transformieren diese Daten ein wenig:

val wordsRDD = textFile.flatMap(line => line.split("\\W"))

Auch hier passiert nichts mit den Daten. Jetzt gibt es eine neue RDD wordsRDD, die einen Verweis auf testFileund eine Funktion enthält, die bei Bedarf angewendet werden kann.

Nur wenn eine Aktion für eine RDD aufgerufen wird, wie z. B. wordsRDD.countdie RDD-Kette, die als Linie bezeichnet wird, wird sie ausgeführt. Das heißt, die in Partitionen aufgeschlüsselten Daten werden von den Ausführenden des Spark-Clusters geladen, die flatMapFunktion wird angewendet und das Ergebnis wird berechnet.

Auf einer linearen Linie wie der in diesem Beispiel cache()wird nicht benötigt. Die Daten werden in die Ausführenden geladen, alle Transformationen werden angewendet und schließlich countwerden die Transformationen berechnet, alle im Speicher - wenn die Daten in den Speicher passen.

cacheist nützlich, wenn sich die Abstammungslinie des RDD verzweigt. Angenommen, Sie möchten die Wörter des vorherigen Beispiels in eine Anzahl für positive und negative Wörter filtern. Sie könnten dies so tun:

val positiveWordsCount = wordsRDD.filter(word => isPositive(word)).count()
val negativeWordsCount = wordsRDD.filter(word => isNegative(word)).count()

Hier gibt jeder Zweig ein erneutes Laden der Daten aus. Durch Hinzufügen einer expliziten cacheAnweisung wird sichergestellt, dass die zuvor durchgeführte Verarbeitung beibehalten und wiederverwendet wird. Der Job wird so aussehen:

val textFile = sc.textFile("/user/emp.txt")
val wordsRDD = textFile.flatMap(line => line.split("\\W"))
wordsRDD.cache()
val positiveWordsCount = wordsRDD.filter(word => isPositive(word)).count()
val negativeWordsCount = wordsRDD.filter(word => isNegative(word)).count()

Aus diesem Grund cachewird gesagt, "die Linie zu brechen", da es einen Prüfpunkt schafft, der für die weitere Verarbeitung wiederverwendet werden kann.

Faustregel: Verwenden cacheSie diese Option , wenn sich die Abstammungslinie Ihres RDD verzweigt oder wenn ein RDD wie in einer Schleife mehrmals verwendet wird.

Müssen wir "Cache" oder "Persist" explizit aufrufen, um die RDD-Daten im Speicher zu speichern?

Ja, nur bei Bedarf.

Die RDD-Daten werden standardmäßig verteilt im Speicher gespeichert?

Nein!

Und das sind die Gründe warum:

• Spark unterstützt zwei Arten von gemeinsam genutzten Variablen: Broadcast-Variablen, mit denen ein Wert im Speicher auf allen Knoten zwischengespeichert werden kann, und Akkumulatoren, bei denen es sich um Variablen handelt, die nur „hinzugefügt“ werden, z. B. Zähler und Summen.

• RDDs unterstützen zwei Arten von Operationen: Transformationen, die ein neues Dataset aus einem vorhandenen erstellen, und Aktionen, die nach dem Ausführen einer Berechnung für das Dataset einen Wert an das Treiberprogramm zurückgeben. Map ist beispielsweise eine Transformation, die jedes Datensatzelement durch eine Funktion leitet und eine neue RDD zurückgibt, die die Ergebnisse darstellt. Auf der anderen Seite ist Reduzieren eine Aktion, die alle Elemente der RDD mithilfe einer Funktion aggregiert und das Endergebnis an das Treiberprogramm zurückgibt (obwohl es auch einen parallelen reduzierten ByKey gibt, der ein verteiltes Dataset zurückgibt).

• Alle Transformationen in Spark sind insofern faul, als sie ihre Ergebnisse nicht sofort berechnen. Stattdessen erinnern sie sich nur an die Transformationen, die auf einen Basisdatensatz (z. B. eine Datei) angewendet wurden. Die Transformationen werden nur berechnet, wenn für eine Aktion ein Ergebnis an das Treiberprogramm zurückgegeben werden muss. Durch dieses Design kann Spark effizienter ausgeführt werden. Beispielsweise können wir erkennen, dass ein über die Karte erstelltes Dataset für eine Reduzierung verwendet wird und nur das Ergebnis der Reduzierung an den Treiber zurückgibt, anstatt das größere zugeordnete Dataset.

• Standardmäßig kann jede transformierte RDD jedes Mal neu berechnet werden, wenn Sie eine Aktion darauf ausführen. Sie können eine RDD jedoch auch mithilfe der Persist- (oder Cache-) Methode im Speicher beibehalten. In diesem Fall behält Spark die Elemente im Cluster bei, um beim nächsten Abfragen einen viel schnelleren Zugriff zu ermöglichen. Es gibt auch Unterstützung für persistente RDDs auf der Festplatte oder für die Replikation über mehrere Knoten.

Weitere Informationen finden Sie in der Spark-Programmieranleitung .

Im Folgenden sind die drei Situationen aufgeführt, in denen Sie Ihre RDDs zwischenspeichern sollten:

viele Male mit einem RDD

Ausführen mehrerer Aktionen auf demselben RDD

für lange Ketten von (oder sehr teuren) Transformationen

Hinzufügen eines weiteren Grundes zum Hinzufügen (oder vorübergehenden Hinzufügen) eines cacheMethodenaufrufs.

für Debug-Speicherprobleme

Mit der cacheMethode gibt spark Debugging-Informationen zur Größe des RDD. In der funkenintegrierten Benutzeroberfläche erhalten Sie Informationen zum RDD-Speicherverbrauch. und dies erwies sich als sehr hilfreich bei der Diagnose von Speicherproblemen.