Habr, salut! Ieri, la , organizat de echipa de la Rambler&Co, au fost multe întrebări din partea participanților legate de configurarea acestui instrument. Am decis să împărtășim experiența noastră pe acest subiect. Tema nu este simplă — așa că vă încurajăm să împărtășiți și voi experiențele în comentarii, poate noi nu înțelegem și nu utilizăm corect.
O mică introducere — cum folosim Spark. Avem un program de trei luni , iar în cadrul celui de-al doilea modul, participanții noștri lucrează cu acest instrument. Prin urmare, sarcina noastră ca organizatori este să pregătim un cluster pentru a fi utilizat în cadrul acestui caz.
Particularitatea utilizării noastre constă în faptul că numărul persoanelor care lucrează simultan cu Spark poate fi egal cu întrega grupă. De exemplu, la seminar, când toată lumea încearcă ceva și repetă după instructorul nostru. Și nu este deloc puțin — uneori aproape 40 de persoane. Probabil nu sunt multe companii în lume care se confruntă cu un astfel de scenariu de utilizare.
Mai apoi, voi explica cum și de ce am ales anumite parametrii pentru configurație.
Să începem cu începutul. Spark poate funcționa în 3 moduri pe un cluster: standalone, folosind Mesos și folosind YARN. Am decis să alegem a treia opțiune, deoarece era logic pentru noi. Avem deja un cluster Hadoop. Participanții noștri sunt deja bine familiarizați cu arhitectura sa. Să folosim YARN.
spark.master=yarnApoi devine mai interesant. Fiecare dintre aceste 3 moduri de implementare are 2 variante de deploy: client și cluster. Bazat pe și diferitele informații de pe internet, se poate concluziona că clientul se potrivește pentru lucrul interactiv — de exemplu, prin jupyter notebook, iar clusterul este mai potrivit pentru soluții de producție. În cazul nostru, ne interesa lucrul interactiv, deci:
spark.deploy-mode=clientÎn esență, de acum înainte, Spark va funcționa pe YARN, dar acest lucru nu era suficient pentru noi. Deoarece lucrăm cu date mari, uneori participanților le lipsea ceea ce se obținea în cadrul împărțirii uniforme a resurselor. Aici am descoperit un element interesant — alocarea dinamică a resurselor. Pe scurt, ideea este următoarea: dacă aveți o sarcină grea și clusterul este liber (de exemplu, dimineața), această opțiune permite Spark-ului să vă ofere resurse suplimentare. Necesitatea este calculată după o formulă complicată. Nu ne vom aprofunda în detalii — funcționează destul de bine.
spark.dynamicAllocation.enabled=trueAm setat acest parametru, iar când am inițiat Spark, acesta a returnat o eroare și nu s-a pornit. Așa e, pentru că ar fi trebuit să citim mai atent. Este indicat că, pentru ca totul să fie în regulă, trebuie să activăm încă un parametru.
spark.shuffle.service.enabled=trueDe ce este necesar? Atunci când jobul nostru nu mai necesită atâtea resurse, Spark ar trebui să le returneze în pool-ul comun. Cea mai consumatoare de timp etapă în aproape orice sarcină MapReduce este etapa Shuffle. Acest parametru permite salvarea datelor care se generează în această etapă, eliberând astfel executors. Iar executorul este procesul care calculează totul pe worker. Acesta are un anumit număr de nuclee de procesor și o anumită cantitate de memorie.
Am adăugat acest parametru. Totul părea să funcționeze. A devenit evident că participanții primeau într-adevăr mai multe resurse atunci când aveau nevoie. Dar a apărut o altă problemă — într-un anumit moment, alți participanți s-au trezit și au dorit să folosească Spark, dar totul era ocupat, iar ei erau nemulțumiți. Ii putem înțelege. Am început să consultăm documentația. Acolo s-a dovedit că există și alte parametere care pot influența procesul. De exemplu, dacă executorul este în modul de așteptare — după cât timp putem lua resursele înapoi?
spark.dynamicAllocation.executorIdleTimeout=120sÎn cazul nostru — dacă executors-urile dvs. nu fac nimic timp de două minute, vă rugăm să le returnați în pool-ul comun. Însă nici acest parametru nu era întotdeauna suficient. Era evident că persoana nu mai făcea nimic de multă vreme, dar resursele nu se eliberau. S-a dovedit că există un alt parametru special — după cât timp să eliminăm executors-urile care conțin date în cache. Implicit, acest parametru era setat pe — infinity! L-am corectat.
spark.dynamicAllocation.cachedExecutorIdleTimeout=600sAdică, dacă executors-urile dvs. nu fac nimic timp de 5 minute, returnați-le în pool-ul comun. În acest mod, viteza de eliberare și alocare a resurselor pentru un număr mare de utilizatori a devenit satisfăcătoare. Numărul de nemulțumiri s-a redus. Dar am decis să mergem mai departe și să limităm numărul maxim de executors pentru o aplicație — practic, pentru un singur participant la program.
spark.dynamicAllocation.maxExecutors=19Acum, desigur, au apărut nemulțumiri din cealaltă parte — “clusterul stă degeaba, iar eu am doar 19 executors”, dar ce să facem — este nevoie de un echilibru corect. Nu se poate face toată lumea fericită.
Și încă o mică poveste legată de specificul cazului nostru. La un moment dat, câțiva oameni au întârziat la exercițiile practice, iar Spark, dintr-un motiv sau altul, nu a pornit. Am verificat cantitatea de resurse disponibile — părea că sunt. Spark ar trebui să pornească. Noroc că, până atunci, documentația reușise să se înregistreze în subconștient și ne-am amintit că, la pornirea Spark, acesta caută un port pe care să pornească. Dacă primul port din interval este ocupat, trece la următorul. Dacă acesta este liber, îl capturează. Și există un parametru care indică numărul maxim de încercări pentru acest lucru. Implicit — este 16. Numărul este mai mic decât numărul de oameni din grupul nostru în timpul lecției. Prin urmare, după 16 încercări, Spark renunța și spunea că nu poate să pornească. Am corectat acest parametru.
spark.port.maxRetries=50Mai departe, voi vorbi despre câteva setări care deja nu sunt foarte legate de specificul cazului nostru.
Pentru o pornire mai rapidă a Spark, există recomandarea de a arhiva folderul jars, aflat în directorul de acasă SPARK_HOME, și de a-l plasa pe HDFS. Astfel, nu va pierde timp descărcând aceste fișiere jar pe workeri.
spark.yarn.archive=hdfs:///tmp/spark-archive.zipDe asemenea, pentru o funcționare mai rapidă, se recomandă utilizarea serializer-ului kryo. Acesta este mai optimizat decât cel implicit.
spark.serializer=org.apache.spark.serializer.KryoSerializerȘi există încă o problemă veche cu Spark, că acesta se oprește adesea din cauza memoriei. De obicei, aceasta se întâmplă în momentul în care lucrătorii au terminat toate calculele și trimit rezultatul către driver. Am crescut acest parametru. Implicit, este de 1GB, noi l-am făcut de 3GB.
spark.driver.maxResultSize=3072Și, în final, ca un bonus. Cum să actualizăm Spark la versiunea 2.1 pe distribuția HortonWorks — HDP 2.5.3.0. Această versiune HDP conține o versiune instalată a 2.0, dar într-o zi am decis că Spark se dezvoltă destul de activ și fiecare versiune nouă repară anumite erori, plus oferă noi funcționalități, inclusiv pentru API-ul Python, așa că am decis că este necesar să facem un update.
Am descărcat versiunea de pe site-ul oficial pentru Hadoop 2.7. Am decomprimat-o, am pus-o în folderul cu HDP. Am creat legături simbolice așa cum trebuie. Am încercat să o lansăm — nu s-a pornit. A dat o eroare foarte confuză.
java.lang.NoClassDefFoundError: com/sun/jersey/api/client/config/ClientConfigCăutând pe Google, am aflat că Spark a decis să nu mai aștepte ca Hadoop să iasă și a decis să folosească o nouă versiune de jersey. Ei se ceartă pe această temă în JIRA. Soluția a fost — să descărcăm . Să o punem în folderul jars în SPARK_HOME, să o zip-uim din nou și să o punem pe HDFS.
Am ocolit această eroare, dar a apărut una nouă și destul de vagă.
org.apache.spark.SparkException: Yarn application has already ended! It might have been killed or unable to launch application masterÎntre timp, am încercat să lansăm versiunea 2.0 — totul a fost ok. Încercă să-ți dai seama care e problema. Ne-am uitat în jurnalele acestei aplicații și am văzut ceva de genul:
/usr/hdp/${hdp.version}/hadoop/lib/hadoop-lzo-0.6.0.${hdp.version}.jarÎn general, din motive necunoscute, hdp.version nu s-a rezolvat. Căutând, am găsit soluția. Trebuie să intri în Ambari în setările YARN și să adaugi acolo parametrul în yarn-site personalizat:
hdp.version=2.5.3.0-37Această magie a ajutat, și Spark a decolat. Am testat câteva dintre jupyter-notebook-urile noastre. Totul funcționează. Suntem pregătiți pentru prima lecție de Spark sâmbătă (deja mâine)!
UPD. La lecție, a apărut o altă problemă. Într-un anumit moment, YARN a încetat să mai acorde containere pentru Spark. A fost nevoie să ajustăm un parametru în YARN, care era implicit 0.2:
yarn.scheduler.capacity.maximum-am-resource-percent=0.8 Asta înseamnă că doar 20% din resurse au fost implicate în distribuția resurselor. După ce am modificat parametrii, am repornit YARN. Problema a fost rezolvată și ceilalți participanți au putut, de asemenea, să pornească contextul Spark.
Sursa: habr.com
