Spark java.lang.OutOfMemoryError: Przestrzeń sterty Java

Mój klaster: 1 master, 11 slave, każdy węzeł ma 6 GB pamięci.

Moje ustawienia:

spark.executor.memory=4g, Dspark.akka.frameSize=512

Oto problem:

Najpierw odczytałem niektóre dane (2,19 GB) z HDFS na RDD:

val imageBundleRDD = sc.newAPIHadoopFile(...)

Po drugie , zrób coś na tym RDD:

val res = imageBundleRDD.map(data => {
                               val desPoints = threeDReconstruction(data._2, bg)
                                 (data._1, desPoints)
                             })

Wreszcie , wyjście do HDFS:

res.saveAsNewAPIHadoopFile(...)

Po uruchomieniu mojego programu pokazuje:

.....
14/01/15 21:42:27 INFO cluster.ClusterTaskSetManager: Starting task 1.0:24 as TID 33 on executor 9: Salve7.Hadoop (NODE_LOCAL)
14/01/15 21:42:27 INFO cluster.ClusterTaskSetManager: Serialized task 1.0:24 as 30618515 bytes in 210 ms
14/01/15 21:42:27 INFO cluster.ClusterTaskSetManager: Starting task 1.0:36 as TID 34 on executor 2: Salve11.Hadoop (NODE_LOCAL)
14/01/15 21:42:28 INFO cluster.ClusterTaskSetManager: Serialized task 1.0:36 as 30618515 bytes in 449 ms
14/01/15 21:42:28 INFO cluster.ClusterTaskSetManager: Starting task 1.0:32 as TID 35 on executor 7: Salve4.Hadoop (NODE_LOCAL)
Uncaught error from thread [spark-akka.actor.default-dispatcher-3] shutting down JVM since 'akka.jvm-exit-on-fatal-error' is enabled for ActorSystem[spark]
java.lang.OutOfMemoryError: Java heap space

Jest zbyt wiele zadań?

PS : Wszystko jest w porządku, gdy dane wejściowe wynoszą około 225 MB.

Jak mogę rozwiązać ten problem?

Mam kilka sugestii:

• Jeśli węzły są skonfigurowane tak, aby mieć 6g maksimum dla Spark (i pozostawiając niewiele dla innych procesów), a następnie użyć 6g zamiast 4G spark.executor.memory=6g. Upewnij się, że używasz jak największej ilości pamięci , sprawdzając interfejs użytkownika (pokaże, ile pamięci używasz)
• Spróbuj użyć większej liczby partycji, powinieneś mieć 2–4 na procesor. IME zwiększenie liczby partycji jest często najłatwiejszym sposobem na uczynienie programu bardziej stabilnym (i często szybszym). W przypadku ogromnych ilości danych może być potrzebnych znacznie więcej niż 4 na procesor, w niektórych przypadkach musiałem użyć 8000 partycji!
• Zmniejsz ułamek pamięci zarezerwowanej do buforowania , używając spark.storage.memoryFraction. Jeśli nie używasz cache()lub nie wpisujesz persistw swoim kodzie, może to również wynosić 0. Domyślnie jest to 0,6, co oznacza, że ​​masz tylko 0,4 * 4 g pamięci na stos. Zmniejszenie IME frac często powoduje, że OOM znikają. AKTUALIZACJA: Od Spark 1.6 najwyraźniej nie będziemy już musieli bawić się tymi wartościami, Spark określi je automatycznie.
• Podobne do powyższego, ale frakcja pamięci losowa . Jeśli twoje zadanie nie wymaga dużej ilości pamięci losowej, ustaw ją na niższą wartość (może to spowodować, że twoje losowe pliki zostaną rozlane na dysk, co może mieć katastrofalny wpływ na szybkość). Czasami, gdy jest to operacja losowania OOM, musisz zrobić coś przeciwnego, tj. Ustawić na coś dużego, na przykład 0,8, lub upewnić się, że pozwalasz, aby tasowanie rozlało się na dysk (jest to domyślne od 1.0.0).
• Uważaj na wycieki pamięci , często są one spowodowane przypadkowym zamknięciem obiektów, których nie potrzebujesz w lambda. Aby zdiagnozować, zwróć uwagę na „zadanie serializowane jako XXX bajtów” w dziennikach, jeśli XXX jest większy niż kilka kilobajtów lub więcej niż MB, możesz mieć przeciek pamięci. Zobacz https://stackoverflow.com/a/25270600/1586965
• Powiązane z powyższym; użyj zmiennych rozgłoszeniowych, jeśli naprawdę potrzebujesz dużych obiektów.
• Jeśli buforujesz duże dyski RDD i możesz poświęcić trochę czasu na dostęp, rozważ szeregowanie RDD http://spark.apache.org/docs/latest/tuning.html#serialized-rdd-storage . Lub nawet buforowanie ich na dysku (co czasem nie jest takie złe, jeśli używasz dysków SSD).
• ( Zaawansowane ) Związane z powyższymi, unikanie Stringi mocno zagnieżdżone struktury (takie jak Mapi zagnieżdżone klasy przypadków). Jeśli to możliwe, staraj się używać tylko typów pierwotnych i indeksować wszystkie elementy inne niż pierwotne, zwłaszcza jeśli oczekujesz wielu duplikatów. Jeśli to możliwe, wybieraj WrappedArraystruktury zagnieżdżone. Lub nawet wdrożyć własną serializację - TY będziesz miał najwięcej informacji na temat efektywnego tworzenia kopii zapasowych danych w bajtach, UŻYJ IT !
• ( nieco zhackowany ) Ponownie, podczas buforowania, rozważ użycie a Datasetdo buforowania struktury, ponieważ będzie ona używać bardziej wydajnej serializacji. Powinno to być traktowane jako hack w porównaniu z poprzednim punktem. Budowanie wiedzy o domenie w algo / serializacji może zminimalizować pamięć / pamięć podręczną o 100x lub 1000x, podczas gdy wszystko, Datasetco prawdopodobnie da, to 2x - 5x w pamięci i 10x skompresowany (parkiet) na dysku.

http://spark.apache.org/docs/1.2.1/configuration.html

EDYCJA: (Więc mogę łatwiej wyszukiwać w Google) Następujące objawy wskazują również na ten problem:

java.lang.OutOfMemoryError : GC overhead limit exceeded

Aby dodać do tego przypadek użycia, który często nie jest omawiany, przedstawię rozwiązanie przy składaniu Sparkwniosku spark-submitw trybie lokalnym .

Według gitbook Mastering Apache Spark przez Jacka Laskowskiego :

Możesz uruchomić Spark w trybie lokalnym. W tym nierozproszonym trybie wdrażania z pojedynczą maszyną JVM Spark spawnuje wszystkie komponenty wykonawcze - sterownik, executor, backend i master - w tej samej maszynie JVM. Jest to jedyny tryb, w którym do wykonania używany jest sterownik.

Zatem jeśli doświadczasz OOM błędy w heap, wystarczy dostosować driver-memoryraczej niż executor-memory.

Oto przykład:

spark-1.6.1/bin/spark-submit
  --class "MyClass"
  --driver-memory 12g
  --master local[*] 
  target/scala-2.10/simple-project_2.10-1.0.jar 

Powinieneś skonfigurować ustawienia pamięci offHeap, jak pokazano poniżej:

val spark = SparkSession
     .builder()
     .master("local[*]")
     .config("spark.executor.memory", "70g")
     .config("spark.driver.memory", "50g")
     .config("spark.memory.offHeap.enabled",true)
     .config("spark.memory.offHeap.size","16g")   
     .appName("sampleCodeForReference")
     .getOrCreate()

Podaj pamięć sterownika i pamięć executora zgodnie z dostępnością pamięci RAM twoich maszyn. Możesz zwiększyć rozmiar offHeap, jeśli nadal masz problem z OutofMemory .

Powinieneś zwiększyć pamięć sterownika. Myślę, że w folderze $ SPARK_HOME / conf powinieneś znaleźć plik spark-defaults.conf, edytować i ustawić spark.driver.memory 4000mzależnie od pamięci w twoim systemie głównym. To właśnie naprawiło problem i wszystko działa płynnie

Spójrz na skrypty startowe, w których ustawiony jest rozmiar sterty Java, wygląda na to, że nie ustawiasz tego przed uruchomieniem robota Spark.

# Set SPARK_MEM if it isn't already set since we also use it for this process
SPARK_MEM=${SPARK_MEM:-512m}
export SPARK_MEM

# Set JAVA_OPTS to be able to load native libraries and to set heap size
JAVA_OPTS="$OUR_JAVA_OPTS"
JAVA_OPTS="$JAVA_OPTS -Djava.library.path=$SPARK_LIBRARY_PATH"
JAVA_OPTS="$JAVA_OPTS -Xms$SPARK_MEM -Xmx$SPARK_MEM"

Dokumentację do wdrażania skryptów można znaleźć tutaj .

Bardzo cierpiałem z powodu tego problemu, używamy dynamicznej alokacji zasobów i myślałem, że wykorzysta zasoby mojego klastra, aby jak najlepiej dopasować aplikację.

Ale prawda jest taka, że ​​dynamiczny przydział zasobów nie ustawia pamięci sterownika i utrzymuje jej domyślną wartość 1 g.

Rozwiązałem go, ustawiając spark.driver.memory na liczbę odpowiadającą pamięci mojego sterownika (dla pamięci RAM 32 GB ustawiłem go na 18 GB)

możesz to ustawić za pomocą polecenia Spark Prześlij w następujący sposób:

spark-submit --conf spark.driver.memory=18gb ....cont

Bardzo ważna uwaga, ta właściwość nie będzie brana pod uwagę, jeśli ustawisz ją z kodu, zgodnie z dokumentacją Spark:

Właściwości Spark można podzielić głównie na dwa rodzaje: jeden jest związany z wdrożeniem, np. „Spark.driver.memory”, „spark.executor.instances”, na tego rodzaju właściwości nie można wpływać przy programowym ustawianiu za pomocą SparkConf w środowisku wykonawczym lub zachowanie zależy od wybranego menedżera klastra i trybu wdrażania, dlatego sugerowane byłoby ustawienie za pomocą pliku konfiguracyjnego lub opcji wiersza polecenia Spark-Submit; inny jest związany głównie z kontrolą środowiska wykonawczego Spark, np. „spark.task.maxFailures”, tego rodzaju właściwości można ustawić w dowolny sposób.

Mówiąc ogólnie, pamięć Spark JVM Executora można podzielić na dwie części. Pamięć Spark i pamięć użytkownika. Jest to kontrolowane przez właściwość spark.memory.fraction- wartość wynosi od 0 do 1. Pracując z obrazami lub wykonując intensywne przetwarzanie pamięci w aplikacjach iskrowych, rozważ zmniejszenie spark.memory.fraction. Dzięki temu więcej pamięci będzie dostępne do pracy aplikacji. Spark może się rozlać, więc nadal będzie działał z mniejszym udziałem pamięci.

Druga część problemu to podział pracy. Jeśli to możliwe, podziel dane na mniejsze części. Mniejsze dane mogą wymagać mniej pamięci. Ale jeśli nie jest to możliwe, poświęcasz obliczenia pamięci. Zazwyczaj pojedynczy moduł wykonujący będzie obsługiwał wiele rdzeni. Całkowita pamięć modułów wykonawczych musi wystarczyć do obsługi wymagań dotyczących pamięci dla wszystkich współbieżnych zadań. Jeśli zwiększenie pamięci modułu wykonującego nie jest możliwe, można zmniejszyć liczbę rdzeni na moduł wykonujący, aby każde zadanie otrzymywało więcej pamięci do pracy. Przetestuj z 1 rdzeniowymi programami wykonawczymi, które mają największą możliwą pamięć, którą możesz dać, a następnie zwiększaj liczbę rdzeni, aż znajdziesz najlepszą liczbę rdzeni.

Zrzuciłeś główny log gc? Więc spotkałem podobny problem i znalazłem, że SPARK_DRIVER_MEMORY ustawia tylko stertę Xmx. Początkowy rozmiar stosu pozostaje 1G, a rozmiar stosu nigdy nie jest skalowany do sterty Xmx.

Przekazywanie „--conf” spark.driver.extraJavaOptions = -Xms20g ”rozwiązuje mój problem.

ps aux | grep java, a zobaczysz następujący dziennik: =

24501 30,7 1,7 41782944 2318184 pkt / 0 Sl + 18:49 0:33 / usr / java / latest / bin / java -cp / opt / spark / conf /: / opt / spark / jars / * -Xmx30g -Xms20g

Miejsce ustawienia wielkości sterty pamięci (przynajmniej w wersji Spark 1.0.0) znajduje się w conf / spark-env. Odpowiednimi zmiennymi są SPARK_EXECUTOR_MEMORY& SPARK_DRIVER_MEMORY. Więcej dokumentów znajduje się w przewodniku wdrażania

Nie zapomnij również skopiować pliku konfiguracyjnego do wszystkich węzłów podrzędnych.

Mam kilka sugestii dotyczących wyżej wspomnianego błędu.

● Sprawdź pamięć modułu wykonującego przypisanego jako moduł wykonujący może mieć do czynienia z partycjami wymagającymi więcej pamięci niż przydzielona.

● Spróbuj sprawdzić, czy więcej przetasowań jest na żywo, ponieważ przetasowania są kosztownymi operacjami, ponieważ obejmują one operacje we / wy dysku, serializację danych i sieciowe operacje we / wy

● Użyj połączeń rozgłoszeniowych

● Unikaj używania groupByKey i spróbuj zamienić na ReduceByKey

● Unikaj używania dużych obiektów Java wszędzie tam, gdzie ma miejsce tasowanie

Z mojego zrozumienia powyższego kodu, ładuje on plik, wykonuje operację mapowania i zapisuje go z powrotem. Nie ma operacji wymagającej odtwarzania losowego. Ponadto, nie ma operacji wymagającej dostarczenia danych do sterownika, dlatego dostrajanie czegokolwiek związanego z tasowaniem lub sterownikiem może nie mieć wpływu. Sterownik ma problemy, gdy jest zbyt wiele zadań, ale było to tylko do wersji Spark 2.0.2. Mogą się zdarzyć dwie rzeczy.

• Jest tylko jeden lub kilka wykonawców. Zwiększ liczbę modułów wykonawczych, aby można je było przypisać do różnych urządzeń podrzędnych. Jeśli używasz przędzy, musisz zmienić konfigurację num-executorów lub jeśli używasz iskier niezależnej, musisz dostroić liczbę rdzeni na executora i conf max rdzeni conf. W standalone num executors = maksymalna liczba rdzeni / rdzeni na moduł wykonujący.
• Liczba partycji jest bardzo niewielka, a może tylko jedna. Więc jeśli jest on niski, nawet jeśli mamy wiele rdzeni, wielu wykonawców, nie będzie to bardzo pomocne, ponieważ równoległość zależy od liczby partycji. Zwiększ partycje, wykonując imageBundleRDD.repartition (11)

Ustawienie tych dokładnych konfiguracji pomogło rozwiązać problem.

spark-submit --conf spark.yarn.maxAppAttempts=2 --executor-memory 10g --num-executors 50 --driver-memory 12g