Spark: Warum übertrifft Python Scala in meinem Anwendungsfall erheblich?

16

Um die Leistung von Spark bei Verwendung von Python und Scala zu vergleichen, habe ich denselben Job in beiden Sprachen erstellt und die Laufzeit verglichen. Ich hatte erwartet, dass beide Jobs ungefähr gleich lange dauern würden, aber der Python-Job dauerte nur 27min, während der Scala-Job dauerte 37min(fast 40% länger!). Ich habe den gleichen Job auch in Java implementiert und es hat auch 37minutesgedauert. Wie ist das möglich, dass Python so viel schneller ist?

Minimales überprüfbares Beispiel:

Python-Job:

# Configuration
conf = pyspark.SparkConf()
conf.set("spark.hadoop.fs.s3a.aws.credentials.provider", "org.apache.hadoop.fs.s3a.AnonymousAWSCredentialsProvider")
conf.set("spark.executor.instances", "4")
conf.set("spark.executor.cores", "8")
sc = pyspark.SparkContext(conf=conf)

# 960 Files from a public dataset in 2 batches
input_files = "s3a://commoncrawl/crawl-data/CC-MAIN-2019-35/segments/1566027312025.20/warc/CC-MAIN-20190817203056-20190817225056-00[0-5]*"
input_files2 = "s3a://commoncrawl/crawl-data/CC-MAIN-2019-35/segments/1566027312128.3/warc/CC-MAIN-20190817102624-20190817124624-00[0-3]*"

# Count occurances of a certain string
logData = sc.textFile(input_files)
logData2 = sc.textFile(input_files2)
a = logData.filter(lambda value: value.startswith('WARC-Type: response')).count()
b = logData2.filter(lambda value: value.startswith('WARC-Type: response')).count()

print(a, b)

Scala Job:

// Configuration
config.set("spark.executor.instances", "4")
config.set("spark.executor.cores", "8")
val sc = new SparkContext(config)
sc.setLogLevel("WARN")
sc.hadoopConfiguration.set("fs.s3a.aws.credentials.provider", "org.apache.hadoop.fs.s3a.AnonymousAWSCredentialsProvider")

// 960 Files from a public dataset in 2 batches 
val input_files = "s3a://commoncrawl/crawl-data/CC-MAIN-2019-35/segments/1566027312025.20/warc/CC-MAIN-20190817203056-20190817225056-00[0-5]*"
val input_files2 = "s3a://commoncrawl/crawl-data/CC-MAIN-2019-35/segments/1566027312128.3/warc/CC-MAIN-20190817102624-20190817124624-00[0-3]*"

// Count occurances of a certain string
val logData1 = sc.textFile(input_files)
val logData2 = sc.textFile(input_files2)
val num1 = logData1.filter(line => line.startsWith("WARC-Type: response")).count()
val num2 = logData2.filter(line => line.startsWith("WARC-Type: response")).count()

println(s"Lines with a: $num1, Lines with b: $num2")

Wenn man sich nur den Code ansieht, scheinen sie identisch zu sein. Ich habe mir die DAGs angesehen und sie haben keine Erkenntnisse geliefert (oder zumindest fehlt mir das Know-how, um eine darauf basierende Erklärung zu finden).

Ich würde mich über Hinweise sehr freuen.

maestromusica
quelle
Kommentare sind nicht für eine ausführliche Diskussion gedacht. Dieses Gespräch wurde in den Chat verschoben .
Samuel Liew
1
Ich hätte die Analyse gestartet, bevor ich etwas gefragt hätte, indem ich die entsprechenden Blöcke und Anweisungen zeitlich festgelegt hätte, um festzustellen, ob es einen bestimmten Ort gibt, an dem die Python-Version schneller ist. Dann hätten Sie vielleicht die Frage schärfen können, warum diese Python-Anweisung schneller ist.
Terry Jan Reedy

Antworten:

11

Ihre Grundannahme, dass Scala oder Java für diese spezielle Aufgabe schneller sein sollten, ist einfach falsch. Sie können dies problemlos mit minimalen lokalen Anwendungen überprüfen. Scala eins:

import scala.io.Source
import java.time.{Duration, Instant}

object App {
  def main(args: Array[String]) {
    val Array(filename, string) = args

    val start = Instant.now()

    Source
      .fromFile(filename)
      .getLines
      .filter(line => line.startsWith(string))
      .length

    val stop = Instant.now()
    val duration = Duration.between(start, stop).toMillis
    println(s"${start},${stop},${duration}")
  }
}

Python eins

import datetime
import sys

if __name__ == "__main__":
    _, filename, string = sys.argv
    start = datetime.datetime.now()
    with open(filename) as fr:
        # Not idiomatic or the most efficient but that's what
        # PySpark will use
        sum(1 for _ in filter(lambda line: line.startswith(string), fr))

    end = datetime.datetime.now()
    duration = round((end - start).total_seconds() * 1000)
    print(f"{start},{end},{duration}")

Ergebnisse (jeweils 300 Wiederholungen, Python 3.7.6, Scala 2.11.12) Posts.xmlaus dem Datendump von hermeneutics.stackexchange.com mit einer Mischung aus übereinstimmenden und nicht übereinstimmenden Mustern:

Boxplots von Durartion in Millis für die oben genannten Programme

  • Python 273,50 (258,84, 288,16)
  • Scala 634,13 (533,81, 734,45)

Wie Sie sehen, ist Python nicht nur systematisch schneller, sondern auch konsistenter (geringere Verbreitung).

Die Nachricht zum Mitnehmen lautet - glauben Sie nicht, dass unbegründete FUD - Sprachen bei bestimmten Aufgaben oder in bestimmten Umgebungen schneller oder langsamer sein können (hier kann Scala beispielsweise vom JVM-Start und / oder GC und / oder JIT getroffen werden), aber wenn Sie dies behaupten wie "XYZ ist X4 schneller" oder "XYZ ist langsam im Vergleich zu ZYX (..) Ungefähr 10x langsamer" bedeutet dies normalerweise, dass jemand wirklich schlechten Code geschrieben hat, um Dinge zu testen.

Bearbeiten :

Um einige in den Kommentaren angesprochene Bedenken auszuräumen:

  • Im OP-Code werden Daten meist in eine Richtung übergeben (JVM -> Python) und es ist keine echte Serialisierung erforderlich (dieser spezifische Pfad wird nur so wie er ist getestet und auf UTF-8 auf der anderen Seite decodiert). Das ist so billig wie es nur geht, wenn es um "Serialisierung" geht.
  • Was zurückgegeben wird, ist nur eine einzelne Ganzzahl pro Partition, sodass die Auswirkungen in dieser Richtung vernachlässigbar sind.
  • Die Kommunikation erfolgt über lokale Steckdosen ( die gesamte Kommunikation auf Arbeitnehmer über die anfänglichen connect und Auth wird durchgeführt unter Verwendung erfolgt Dateideskriptors von zurückgegeben local_connect_and_auth, und es ist nichts anderes als Socket - Datei ). Wieder so billig wie es nur geht, wenn es um die Kommunikation zwischen Prozessen geht.
  • In Anbetracht des oben gezeigten Unterschieds in der Rohleistung (viel höher als in Ihrem Programm angegeben) gibt es einen großen Spielraum für die oben aufgeführten Gemeinkosten.
  • Dieser Fall unterscheidet sich grundlegend von Fällen, in denen entweder einfache oder komplexe Objekte in einer Form an und von Python-Interpreter übergeben werden müssen, die beiden Parteien als Pickle-kompatible Dumps zugänglich ist (die bemerkenswertesten Beispiele sind UDF im alten Stil, einige Teile alter MLLib).

Bearbeiten 2 :

Da Jasper-m hier über die Startkosten besorgt war, kann man leicht beweisen, dass Python gegenüber Scala immer noch einen erheblichen Vorteil hat, selbst wenn die Eingabegröße erheblich erhöht wird.

Hier sind die Ergebnisse für 2003360 Zeilen / 5.6G (dieselbe Eingabe, nur mehrfach dupliziert, 30 Wiederholungen), die alles übertreffen, was Sie von einer einzelnen Spark-Aufgabe erwarten können.

Geben Sie hier die Bildbeschreibung ein

  • Python 22809.57 (21466.26, 24152.87)
  • Scala 27315.28 (24367.24, 30263.31)

Bitte beachten Sie nicht überlappende Konfidenzintervalle.

Edit 3 :

Um einen weiteren Kommentar von Jasper-M anzusprechen :

Der Großteil der gesamten Verarbeitung findet im Spark-Fall noch in einer JVM statt.

Das ist in diesem speziellen Fall einfach falsch:

  • Bei dem fraglichen Job handelt es sich um einen Kartenjob mit einer einzelnen globalen Reduzierung mithilfe von PySpark-RDDs.
  • PySpark RDD implementiert (anders als zum Beispiel DataFrame) eine Vielzahl von Funktionen nativ in Python, mit Ausnahme der Eingabe, Ausgabe und Kommunikation zwischen Knoten.
  • Da es sich um einen einstufigen Job handelt und die endgültige Ausgabe klein genug ist, um ignoriert zu werden, besteht die Hauptverantwortung von JVM (wenn man nicht pickt, wird dies hauptsächlich in Java und nicht in Scala implementiert) darin, das Hadoop-Eingabeformat aufzurufen und Daten über den Socket zu übertragen Datei zu Python.
  • Der gelesene Teil ist für die JVM- und Python-API identisch, sodass er als konstanter Overhead betrachtet werden kann. Es ist auch nicht der Hauptteil der Verarbeitung , selbst für solch einfache Aufgaben wie diese.
user10938362
quelle
3
ausgezeichnete Herangehensweise an das Problem. Vielen Dank für das Teilen
Alexandros Biratsis
1
@egordoe Alexandros sagte, "hier wird keine UDF aufgerufen", nicht "Python wird nicht aufgerufen" - das macht den Unterschied. Der Serialisierungsaufwand ist wichtig, wenn Daten zwischen Systemen ausgetauscht werden (dh wenn Sie Daten an eine UDF und zurück übergeben möchten).
user10938362
1
@egordoe Sie verwechseln eindeutig zwei Dinge - den Overhead der Serialisierung, der ein Problem darstellt, wenn nicht triviale Objekte hin und her übergeben werden. Und Overhead der Kommunikation. Hier gibt es wenig oder keinen Serialisierungsaufwand, da Sie nur Bytestrings übergeben und dekodieren. Dies geschieht meistens in Richtung, da Sie zurück eine einzelne Ganzzahl pro Partition erhalten. Die Kommunikation ist besorgniserregend, aber die Weitergabe von Daten über lokale Sockets ist effizient, da dies bei der Kommunikation zwischen Prozessen tatsächlich der Fall ist. Wenn das nicht klar ist, empfehle ich, die Quelle zu lesen - es ist nicht schwer und wird aufschlussreich sein.
user10938362
1
Außerdem werden Serialisierungsmethoden einfach nicht gleichgestellt. Wie der Spark-Fall zeigt, können gute Serialisierungsmethoden die Kosten auf ein Niveau senken, auf dem sie nicht mehr von Bedeutung sind (siehe Pandas UDF mit Pfeil), und wenn dies geschieht, können andere Faktoren dominieren (siehe beispielsweise Leistungsvergleiche zwischen Scala-Fensterfunktionen und deren Entsprechungen mit Pandas) UDFs - Python gewinnt dort mit viel höherem Vorsprung als in dieser Frage.
user10938362
1
Und dein Punkt ist @ Jasper-M? Einzelne Spark-Aufgaben sind normalerweise klein genug, um eine vergleichbare Arbeitslast zu haben. Verstehen Sie mich nicht falsch, aber wenn Sie ein Gegenbeispiel haben, das entweder diese oder die gesamte Frage ungültig macht, posten Sie es bitte. Ich habe bereits festgestellt, dass sekundäre Maßnahmen in gewissem Maße zu diesem Wert beitragen, aber die Kosten nicht dominieren. Wir sind alle Ingenieure (irgendeiner Art) hier - lassen Sie uns über Zahlen und Code sprechen, nicht über Überzeugungen, oder?
user10938362
4

Der Scala-Job dauert länger, da er falsch konfiguriert ist und daher die Python- und Scala-Jobs mit ungleichen Ressourcen ausgestattet wurden.

Der Code enthält zwei Fehler:

val sc = new SparkContext(config) // LINE #1
sc.setLogLevel("WARN")
sc.hadoopConfiguration.set("fs.s3a.aws.credentials.provider", "org.apache.hadoop.fs.s3a.AnonymousAWSCredentialsProvider")
sc.hadoopConfiguration.set("spark.executor.instances", "4") // LINE #4
sc.hadoopConfiguration.set("spark.executor.cores", "8") // LINE #5
  1. LINE 1. Sobald die Zeile ausgeführt wurde, ist die Ressourcenkonfiguration des Spark-Jobs bereits eingerichtet und festgelegt. Ab diesem Zeitpunkt kann nichts mehr angepasst werden. Weder die Anzahl der Executoren noch die Anzahl der Kerne pro Executor.
  2. Zeile 4-5. sc.hadoopConfigurationist ein falscher Ort, um eine Spark-Konfiguration festzulegen. Es sollte in der configInstanz festgelegt werden, an die Sie übergeben new SparkContext(config).

[HINZUGEFÜGT] Vor diesem Hintergrund würde ich vorschlagen, den Code des Scala-Jobs in zu ändern

config.set("spark.executor.instances", "4")
config.set("spark.executor.cores", "8")
val sc = new SparkContext(config) // LINE #1
sc.setLogLevel("WARN")
sc.hadoopConfiguration.set("fs.s3a.aws.credentials.provider", "org.apache.hadoop.fs.s3a.AnonymousAWSCredentialsProvider")

und testen Sie es erneut. Ich wette, die Scala-Version wird jetzt X-mal schneller sein.

egordoe
quelle
Ich habe überprüft, dass beide Jobs 32 Aufgaben parallel ausführen, sodass ich nicht glaube, dass dies der Schuldige ist.
Maestromusica
danke für die Bearbeitung, werde versuchen, es jetzt zu testen
maestromusica
hi @maestromusica, es muss etwas in der Ressourcenkonfiguration sein, da Python Scala in diesem speziellen Anwendungsfall möglicherweise nicht übertrifft. Ein weiterer Grund können einige nicht korrelierte Zufallsfaktoren sein, z. B. die Belastung des Clusters zum jeweiligen Zeitpunkt und ähnliches. Übrigens, welchen Modus benutzt du? Standalone, lokal, Garn?
Egordoe
Ja, ich habe überprüft, dass diese Antwort falsch ist. Die Laufzeit ist die gleiche. Ich habe in beiden Fällen auch die Konfiguration gedruckt und sie ist identisch.
Maestromusica
1
Ich denke du hast vielleicht recht. Ich habe diese Frage gestellt, um alle anderen Möglichkeiten zu untersuchen, wie z. B. einen Fehler im Code oder vielleicht, dass ich etwas falsch verstanden habe. Danke für deinen Beitrag.
Maestromusica