pyspark Cache

Question

hier nicht zu beschleunigen scheint ich pyspark bin mit einem einfachen wordcount zu tun, habe ich den Cache-Methode, aber scheint nicht das zweite Mal zu beschleunigen betreibe ich den Code:

from pyspark import SparkContext 
from time import time 
if __name__=='__main__': 
    sc=SparkContext() 
    rdd=sc.textFile("shakespear.txt") 
    rdd.cache() 
    t1=time() 
    rdd.flatMap(lambda r:r.split(" "))\ 
     .map(lambda w:(w,1))\ 
     .reduceByKey(lambda x,y:x+y)\ 
     .collect() 
    t2=time() 
    print t2-t1 
    t3=time() 
    rdd.flatMap(lambda r:r.split(" "))\ 
     .map(lambda w:(w,1))\ 
     .reduceByKey(lambda x,y:x+y)\ 
     .collect() 
    t4=time() 
    print t4-t3 

dann ist das Ergebnis:

3.87242698669 
2.49692296982 

Wenn ich die rdd.cache Kommentar aus(), das Ergebnis ist:

3.58677482605 
2.52048921585 

Beim Lesen des Funken tun Ich habe gehört, dass Cache den Prozess sehr beschleunigen würde, richtig? Aber hier macht es keinen großen Unterschied, könnte mir jemand einen Hinweis auf dieses spezielle Thema geben? Eine allgemeine Einführung in die Funktionsweise von Spark-Cache wäre großartig! Vielen Dank!

Answer 1

Diese Textdatei ist wahrscheinlich klein genug, dass Spark sie implizit während der Sitzung zwischenspeichert. Es gibt also keinen großen Unterschied zwischen zwei Läufen und auch deshalb ist der zweite Aufruf der rdd in beiden Läufen etwas schneller.

Answer 2

Zuerst ruft cache nichts berechnet. Es gibt nur an, wenn eine RDD berechnet wird, werden die Ergebnisse zwischengespeichert. Die Berechnung einer RDD wird nur durch Aktionen wie collect ausgelöst.

Caching kann jedoch nur verwendet werden, wenn beim zweiten Zugriff der gleiche Verweis auf die RDD verwendet wird. Zum Beispiel, hier ist eine Version des Programms über, wo die zweite Berechnung tatsächlich Verwendung des Cache machen:

from pyspark import SparkContext 
from time import time 
if __name__=='__main__': 
    sc=SparkContext() 
    rdd=sc.textFile("shakespear.txt") 
    rdd.cache() 
    t1=time() 
    rdd2 = rdd.flatMap(lambda r:r.split(" "))\ 
       .map(lambda w:(w,1))\ 
       .reduceByKey(lambda x,y:x+y)\ 
    rdd2.collect() 
    t2=time() 
    print t2-t1 
    t3=time() 
    rdd2.collect() 
    t4=time() 
    print t4-t3 

Bitte beachte, dass ich eine neue Variable erstellt rdd2 benannt, die die Transformationen definiert, die Sie neu berechnen möchten.