Spark: Dlaczego Python znacznie przewyższa Scalę w moim przypadku użycia?

Aby porównać wydajność Spark przy użyciu Pythona i Scali, stworzyłem to samo zadanie w obu językach i porównałem środowisko wykonawcze. Spodziewałem się, że obie prace zajmą mniej więcej tyle samo czasu, ale praca w Pythonie trwała tylko 27min, podczas gdy praca w Scali trwała 37min(prawie 40% dłużej!). Zaimplementowałem również tę samą pracę w Javie i to 37minutesteż zajęło . Jak to możliwe, że Python jest o wiele szybszy?

Minimalny możliwy do zweryfikowania przykład:

Zadanie Python:

# 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)

Praca Scala:

// 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")

Po prostu patrząc na kod wydają się identyczne. Spojrzałem na DAG i nie dostarczyli oni żadnych spostrzeżeń (a przynajmniej brakuje mi know-how, aby znaleźć na ich podstawie wyjaśnienie).

Byłbym naprawdę wdzięczny za wszelkie wskazówki.

Twoje podstawowe założenie, że Scala lub Java powinny być szybsze do tego konkretnego zadania, jest po prostu błędne. Możesz to łatwo zweryfikować za pomocą minimalnych aplikacji lokalnych. Scala One:

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 pierwszy

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}")

Wyniki (300 powtórzeń każda, Python 3.7.6, Scala 02.11.12) w sprawie Posts.xmlz hermeneutics.stackexchange.com dane zrzucić z mieszanką i dopasowywania wzorców non pasujące:

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

Jak widzisz, Python jest nie tylko systematycznie szybszy, ale także bardziej spójny (mniejszy spread).

Komunikat „zabierz” to - nie wierz w bezpodstawny FUD - języki mogą być szybsze lub wolniejsze w określonych zadaniach lub w określonych środowiskach (na przykład tutaj Scala może zostać uderzona przez uruchomienie JVM i / lub GC i / lub JIT), ale jeśli twierdzisz, że jak „XYZ jest X4 szybszy” lub „XYZ jest wolny w porównaniu do ZYX (..) Około, 10 razy wolniejszy” zwykle oznacza to, że ktoś napisał naprawdę zły kod do testowania.

Edytuj :

Aby zaradzić niektórym obawom wyrażonym w komentarzach:

• W kodzie OP dane są przekazywane głównie w jednym kierunku (JVM -> Python) i nie jest wymagana żadna prawdziwa serializacja (ta konkretna ścieżka po prostu przechodzi przez testowanie bez zmian i dekodowanie na UTF-8 po drugiej stronie). Jest to tak tanie, jak to tylko możliwe, jeśli chodzi o „serializację”.
• To, co jest przekazywane, to tylko jedna liczba całkowita według partycji, więc wpływ w tym kierunku jest znikomy.
• Komunikacja odbywa się za pośrednictwem lokalnych gniazd (cała komunikacja w ramach procesu roboczego poza początkowym połączeniem i uwierzytelnianiem odbywa się za pomocą zwróconego deskryptora plikulocal_connect_and_auth i jest niczym innym jak plikiem powiązanym z gniazdem ). Ponownie, tak tanie, jak to możliwe, jeśli chodzi o komunikację między procesami.
• Biorąc pod uwagę różnicę w wydajności pierwotnej pokazaną powyżej (znacznie wyższą niż to, co widzisz w swoim programie), istnieje duży margines dla kosztów ogólnych wymienionych powyżej.
• Ten przypadek różni się całkowicie od przypadków, w których proste lub złożone obiekty muszą być przekazywane do iz interpretera Pythona w formie, która jest dostępna dla obu stron jako zrzuty kompatybilne z zalewami (najbardziej znaczącymi przykładami są UDF w starym stylu, niektóre części starego MLLib).

Edycja 2 :

Ponieważ jasper-m był zaniepokojony kosztem uruchomienia tutaj, łatwo można udowodnić, że Python nadal ma znaczącą przewagę nad Scalą, nawet jeśli rozmiar wejściowy zostanie znacznie zwiększony.

Oto wyniki dla 2003360 linii / 5.6G (to samo wejście, po prostu zduplikowane wiele razy, 30 powtórzeń), który to sposób przekracza wszystko, czego można oczekiwać w jednym zadaniu Spark.

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

Uwaga: nie nakładające się przedziały ufności.

Edycja 3 :

Aby odpowiedzieć na inny komentarz Jasper-M:

Większość przetwarzania nadal odbywa się w JVM w przypadku Spark.

Jest to po prostu niepoprawne w tym konkretnym przypadku:

• Zadanie, o którym mowa, to zadanie mapy z pojedynczą globalną redukcją przy użyciu RDD PySpark.
• PySpark RDD (inaczej niż powiedzmy DataFrame) implementuje ogromną funkcjonalność natywnie w Pythonie, z wyjątkami wejścia, wyjścia i komunikacji między węzłami.
• Ponieważ jest to jednoetapowe zadanie, a końcowy wynik jest wystarczająco mały, aby go zignorować, głównym obowiązkiem JVM (jeśli ktoś ma nitpick, jest to implementowane głównie w Javie, a nie Scali) jest wywoływanie formatu wejściowego Hadoop i przepychanie danych przez gniazdo plik do Pythona.
• Część do odczytu jest identyczna dla JVM i API Pythona, więc można ją uznać za stały narzut. Nie kwalifikuje się również jako większość przetwarzania , nawet w przypadku tak prostej pracy jak ta.

Zadanie Scala trwa dłużej, ponieważ ma błędną konfigurację, dlatego zadania Python i Scala zostały wyposażone w nierówne zasoby.

W kodzie są dwa błędy:

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. LINIA 1. Po wykonaniu linii konfiguracja zasobów zadania Spark jest już ustalona i naprawiona. Od tego momentu nic nie można zmienić. Ani liczba wykonawców, ani liczba rdzeni na wykonawcę.
2. LINIA 4-5. sc.hadoopConfigurationjest złym miejscem do ustawienia dowolnej konfiguracji Spark. Powinien być ustawiony w configinstancji, którą przekazujesz new SparkContext(config).

[DODANO] Mając powyższe na uwadze, proponuję zmienić kod zadania Scala na

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")

i ponownie go przetestuj. Założę się, że wersja Scali będzie teraz X razy szybsza.