Cześć Habr! Wczoraj na , organizowanym przez chłopaków z Rambler&Co, było dość wiele pytań od uczestników dotyczących konfigurowania tego narzędzia. Postanowiliśmy podzielić się naszym doświadczeniem. Temat jest skomplikowany — dlatego zachęcamy do dzielenia się własnymi doświadczeniami w komentarzach, być może również mamy swoje nieporozumienia.
Krótki wstęp — jak używamy Sparka. Mamy trzy miesięczny program , a w całym drugim module nasi uczestnicy pracują z tym narzędziem. Naszym zadaniem jako organizatorów jest przygotowanie klastra do pracy w przypadku takiego użycia.
Specyfika naszego użycia polega na tym, że liczba osób jednocześnie pracujących na Sparku może być równa całej grupie. Na przykład, podczas seminarium, kiedy wszyscy jednocześnie coś próbują i powtarzają za naszym nauczycielem. A to nie jest mało — czasami pod 40 osób. Pewnie nie ma zbyt wielu firm na świecie, które stają przed takim scenariuszem użycia.
Później opiszę, jak i dlaczego dobieraliśmy poszczególne parametry konfiguracji.
Zacznijmy od samego początku. Spark ma 3 opcje pracy na klastrze: standalone, z użyciem Mesos i z użyciem YARN. Postanowiliśmy wybrać trzecią opcję, ponieważ była dla nas najbardziej logiczna. Mamy już klaster hadoop. Nasi uczestnicy są dobrze zaznajomieni z jego architekturą. Użyjmy YARN.
spark.master=yarnDalej robi się ciekawiej. Każda z tych 3 opcji wdrożenia ma 2 opcje klasyfikacji: client i cluster. Na podstawie i różnych linków w internecie można wywnioskować, że client nadaje się do interaktywnej pracy — na przykład przez jupyter notebook, a cluster bardziej nadaje się do rozwiązań produkcyjnych. W naszym przypadku interesowała nas interaktywna praca, więc:
spark.deploy-mode=clientW zasadzie od tego momentu Spark zaczął jakoś działać na YARN, ale to nam nie wystarczyło. Ponieważ pracujemy z dużymi danymi, czasami uczestnikom brakowało tego, co można było uzyskać w ramach równomiernego podziału zasobów. I tutaj odkryliśmy ciekawą rzecz – dynamiczną alokację zasobów. Krótko mówiąc, gdy masz ciężkie zadanie i klaster jest wolny (na przykład rano), dzięki tej opcji Spark może przyznać ci dodatkowe zasoby. Wymagania obliczane są na podstawie skomplikowanej formuły. Nie będziemy wchodzić w szczegóły — działa to całkiem nieźle.
spark.dynamicAllocation.enabled=trueUstawiliśmy ten parametr, a przy uruchomieniu Spark zgłosił błąd i się nie uruchomił. Słusznie, bo trzeba było przeczytać uważniej. Jest tam napisane, że aby wszystko było w porządku, należy dodatkowo włączyć jeszcze jeden parametr.
spark.shuffle.service.enabled=truePo co to jest potrzebne? Kiedy nasze zadanie przestaje wymagać takiej ilości zasobów, Spark powinien zwrócić je do ogólnej puli. Najbardziej czasochłonna faza w prawie każdym zadaniu MapReduce to faza Shuffle. Ten parametr pozwala zachować dane, które powstają w tej fazie, a tym samym zwolnić executory. A executor — to proces, który na węźle roboczym oblicza wszystko. Ma określoną liczbę rdzeni procesora i określoną ilość pamięci.
Dodaliśmy ten parametr. Wszystko wydaje się działać. Zauważono, że uczestnikom naprawdę zaczęło się przydzielać więcej zasobów, kiedy ich potrzebowali. Ale pojawił się inny problem — w pewnym momencie inni uczestnicy się budzili i też chcieli korzystać z Sparka, a tam wszystko było zajęte, więc byli niezadowoleni. Można ich zrozumieć. Zaczęliśmy przeglądać dokumentację. Okazało się, że jest jeszcze kilka parametrów, które mogą wpłynąć na ten proces. Na przykład, jeśli executor jest w trybie oczekiwania — po jakim czasie można zabrać mu zasoby?
spark.dynamicAllocation.executorIdleTimeout=120sW naszym przypadku — jeśli twoje executory nic nie robią przez dwie minuty, to proszę, zwróć je do wspólnej puli. Ale nawet ten parametr nie zawsze wystarczył. Było widać, że ktoś od dawna nic nie robi, a zasoby nie są zwalniane. Okazało się, że istnieje jeszcze jeden specjalny parametr — po jakim czasie należy odebrać executory, które zawierają dane w pamięci podręcznej. Domyślnie ten parametr był ustawiony na — nieskończoność! Naprawiliśmy to.
spark.dynamicAllocation.cachedExecutorIdleTimeout=600sTo znaczy, że jeśli w ciągu 5 minut twoje executory nic nie robią, oddaj je do wspólnej puli. W takim trybie prędkość zwalniania i dostarczania zasobów dla dużej liczby użytkowników stała się odpowiednia. Liczba niezadowolonych zmniejszyła się. Ale postanowiliśmy pójść dalej i ograniczyć maksymalną liczbę executorów na jedną aplikację — w zasadzie na jednego uczestnika programu.
spark.dynamicAllocation.maxExecutors=19Teraz, oczywiście, pojawiły się niezadowolone osoby z drugiej strony — „klaster stoi bezczynnie, a mam tylko 19 executorów”, ale cóż zrobić — potrzebny jest jakiś odpowiedni balans. Nie da się wszystkich uszczęśliwić.
I jeszcze jedna mała historia związana ze specyfiką naszego przypadku. Kiedyś na zajęcia praktyczne spóźniło się kilka osób, i ich Spark z jakiegoś powodu nie wystartował. Sprawdziliśmy liczbę dostępnych zasobów — wydawało się, że są. Spark powinien wystartować. Na szczęście, do tego czasu dokumentacja już gdzieś zapisała się w podświadomości i przypomnieliśmy sobie, że przy uruchamianiu Spark szuka portu, na którym może wystartować. Jeśli pierwszy port z zakresu jest zajęty, przechodzi do następnego. Jeśli jest wolny, to go zajmuje. I jest parametr, który wskazuje maksymalną liczbę prób dla tego. Domyślnie — to 16. Liczba ta jest mniejsza niż liczba osób w naszej grupie na zajęciach. W związku z tym, po 16 próbach Spark rezygnował i mówił, że nie może wystartować. Naprawiliśmy ten parametr.
spark.port.maxRetries=50Teraz opowiem o niektórych ustawieniach, które nie są już zbytnio związane ze specyfiką naszego przypadku.
Aby szybciej uruchomić Spark, zaleca się skompresować folder jars znajdujący się w domowym katalogu SPARK_HOME i umieścić na HDFS. W ten sposób nie będzie tracić czasu na ładowanie tych jarów na węzły robocze.
spark.yarn.archive=hdfs:///tmp/spark-archive.zipZalecamy też używanie kryo jako serializera, aby przyspieszyć pracę. Jest bardziej zoptymalizowany niż ten domyślny.
spark.serializer=org.apache.spark.serializer.KryoSerializerIstnieje też od dawna problem z Spark, że często pada z powodu braku pamięci. Zwykle dzieje się to w momencie, gdy pracownicy wszystko obliczą i wysyłają wyniki do sterownika. Ustawiliśmy ten parametr na wyższą wartość. Domyślnie wynosi 1GB, a my ustawiliśmy na 3.
spark.driver.maxResultSize=3072I na koniec, jako deser. Jak zaktualizować Spark do wersji 2.1 na dystrybucji HortonWorks — HDP 2.5.3.0. Ta wersja HDP zawiera zainstalowaną wersję 2.0, ale pewnego razu postanowiliśmy, że Spark rozwija się dość aktywnie, a każda nowa wersja naprawia jakieś błędy i oferuje dodatkowe możliwości, w tym dla python API, więc postanowiliśmy, że trzeba przeprowadzić aktualizację.
Pobraliśmy wersję z oficjalnej strony pod Hadoop 2.7. Wypakowaliśmy, wrzuciliśmy do folderu HDP. Ustawiliśmy symlinki jak trzeba. Uruchamiamy — nie startuje. Wyświetla bardzo niezrozumiały błąd.
java.lang.NoClassDefFoundError: com/sun/jersey/api/client/config/ClientConfigPo googlowaniu dowiedzieliśmy się, że Spark postanowił nie czekać, aż Hadoop zostanie zaktualizowany, i zdecydował się na nową wersję jersey. Oni sami się tam kłócą na ten temat w JIRA. Rozwiązaniem było pobranie . Wrzucamy to do folderu jars w SPARK_HOME, ponownie robimy zip i wrzucamy na HDFS.
Obeszliśmy ten błąd, ale pojawił się nowy, dość niejasny.
org.apache.spark.SparkException: Yarn application has already ended! It might have been killed or unable to launch application masterPróbując uruchomić wersję 2.0 — wszystko jest w porządku. Spróbuj zgadnąć, o co chodzi. Zajrzelibyśmy do logów tej aplikacji i zobaczyliśmy coś takiego:
/usr/hdp/${hdp.version}/hadoop/lib/hadoop-lzo-0.6.0.${hdp.version}.jarOgółem, z jakiegoś powodu hdp.version nie została rozwiązana. Po googlowaniu znaleźliśmy rozwiązanie. Trzeba wejść w Ambari w ustawienia YARN i dodać parametr w custom yarn-site:
hdp.version=2.5.3.0-37Ta magia zadziałała i Spark ruszył. Przetestowaliśmy kilka naszych notebooków Jupyter. Wszystko działa. Jesteśmy gotowi na pierwsze zajęcia z Sparka w sobotę (już jutro)!
UPD. Na zajęciach wyszła jeszcze jedna sprawa. W pewnym momencie YARN przestał przydzielać kontenery dla Sparka. W YARN trzeba było poprawić parametr, który domyślnie wynosił 0.2:
yarn.scheduler.capacity.maximum-am-resource-percent=0.8 To znaczy, tylko 20% zasobów brało udział w przydzielaniu zasobów. Zmieniając parametry, zrestartowano YARN. Problem został rozwiązany, a pozostali uczestnicy również mogli uruchomić kontekst Spark.
Źródło: habr.com
