Ciao Habr! Ieri durante , organizzato dai ragazzi di Rambler&Co, ci sono state molte domande dai partecipanti riguardo alla configurazione di questo strumento. Abbiamo deciso di condividere la nostra esperienza a riguardo. La tematica non è semplice, quindi ci piacerebbe che anche voi condivideste le vostre esperienze nei commenti, magari ci sono aspetti che non comprendiamo o utilizziamo correttamente.
Una piccola introduzione: come utilizziamo Spark. Abbiamo un programma di tre mesi , e nel secondo modulo i nostri partecipanti lavorano con questo strumento. Di conseguenza, il nostro compito come organizzatori è preparare un cluster per l'uso in questo caso specifico.
La particolarità del nostro utilizzo è che il numero di persone che lavorano contemporaneamente su Spark può corrispondere all'intero gruppo. Ad esempio, durante il seminario, quando tutti provano e ripetono ciò che fa il nostro docente. E questo non sono pochi: talvolta quasi 40 persone. Probabilmente ci sono poche aziende al mondo che affrontano uno scenario di utilizzo simile.
In seguito vi spiegherò come e perché abbiamo scelto determinati parametri di configurazione.
Iniziamo dall'inizio. Spark offre 3 modalità per operare su un cluster: standalone, utilizzando Mesos e con YARN. Abbiamo deciso di optare per la terza opzione, poiché era la più logica per noi. Abbiamo già un cluster Hadoop. I nostri partecipanti conoscono bene la sua architettura. Utilizziamo YARN.
spark.master=yarnOra diventa interessante. Ognuna di queste 3 modalità di distribuzione ha 2 opzioni di deploy: client e cluster. Basandoci su e su vari riferimenti su internet, si può dedurre che la modalità client è adatta per il lavoro interattivo, ad esempio tramite jupyter notebook, mentre la modalità cluster è più adatta per soluzioni in produzione. Nel nostro caso, ci interessava il lavoro interattivo, quindi:
spark.deploy-mode=clientIn effetti, da questo momento Spark inizierà a funzionare su YARN, ma non era sufficiente per noi. Poiché ci occupiamo di big data, a volte ai partecipanti mancavano le risorse che venivano fornite tramite una distribuzione uniforme. E qui abbiamo trovato qualcosa di interessante: l'allocazione dinamica delle risorse. In breve, la questione è la seguente: se hai un compito pesante e il cluster è libero (per esempio, al mattino), grazie a questa opzione Spark può fornirti risorse aggiuntive. La necessità viene calcolata tramite una formula piuttosto astuta. Non entriamo nei dettagli, funziona piuttosto bene.
spark.dynamicAllocation.enabled=trueAbbiamo impostato questo parametro, e all'avvio Spark ha sollevato un errore e non si è avviato. È giusto, perché avremmo dovuto leggere più attentamente. È specificato che per far funzionare tutto correttamente, è necessario abilitare un ulteriore parametro.
spark.shuffle.service.enabled=trueA cosa serve? Quando il nostro job non richiede più una certa quantità di risorse, Spark deve restituirle al pool generale. La fase più laboriosa in quasi qualsiasi compito MapReduce è la fase di Shuffle. Questo parametro consente di conservare i dati generati in questa fase e di liberare gli executors di conseguenza. Un executor è un processo che calcola tutto sul worker. Ha un certo numero di core e una certa quantità di memoria.
Abbiamo aggiunto questo parametro. Tutto sembra funzionare. È diventato evidente che ai partecipanti venivano realmente forniti più risorse quando ne avevano bisogno. Ma è emerso un altro problema: a un certo punto, gli altri partecipanti si svegliavano e volevano anche utilizzare Spark, ma tutto era occupato, e loro erano scontenti. Possono capire. Abbiamo iniziato a guardare nella documentazione. Lì abbiamo scoperto che ci sono ancora alcuni parametri tramite i quali è possibile influenzare il processo. Ad esempio, se l'executor è in modalità inattesa, dopo quanto tempo possiamo recuperare le risorse?
spark.dynamicAllocation.executorIdleTimeout=120sNel nostro caso, se i vostri executors non fanno nulla per due minuti, vi preghiamo di restituirli al pool comune. Ma anche questo parametro non era sempre sufficiente. Era evidente che una persona non stava facendo nulla da tempo, ma le risorse non venivano rilasciate. Si è scoperto che c'era un altro parametro speciale: dopo quanto tempo ritirare gli executors che contengono dati cache. Di default, questo parametro era impostato su infinity! Lo abbiamo corretto.
spark.dynamicAllocation.cachedExecutorIdleTimeout=600sQuindi, se per 5 minuti i vostri executors non fanno nulla, consegnateli al pool comune. In questo modo, la velocità di rilascio e distribuzione delle risorse per un gran numero di utenti è diventata soddisfacente. Il numero di lamentele è diminuito. Ma abbiamo deciso di andare oltre e limitare il numero massimo di executors per un'applicazione, essenzialmente per un partecipante al programma.
spark.dynamicAllocation.maxExecutors=19Ora, naturalmente, ci sono stati dei reclami dall'altro lato: “il cluster è inattivo e ho solo 19 executors”, ma che ci possiamo fare — è necessario trovare un giusto equilibrio. Non è possibile rendere tutti felici.
E un'altra piccola storia legata alle specificità del nostro caso. Un giorno, durante una lezione pratica, alcune persone sono arrivate in ritardo e per qualche motivo Spark non è partito. Abbiamo controllato la quantità di risorse disponibili — sembrava ci fossero. Spark dovrebbe avviarsi. Fortunatamente, a quel punto la documentazione era già registrata nella nostra memoria e ci siamo ricordati che, quando avvia Spark, cerca una porta sulla quale partire. Se la prima porta dell'intervallo è occupata, passa a quella successiva. Se è libera, la prende. E c'è un parametro che indica il numero massimo di tentativi per questo. Per impostazione predefinita, è 16. Un numero inferiore al numero di persone nel nostro gruppo durante la lezione. Di conseguenza, dopo 16 tentativi, Spark abbandonava e diceva che non poteva avviarsi. Abbiamo corretto questo parametro.
spark.port.maxRetries=50Ora parlerò di alcune impostazioni che non sono molto legate alla specificità del nostro caso.
Per un avvio più rapido di Spark, si consiglia di comprimere la cartella jars, situata nella directory home di SPARK_HOME, e di caricarla su HDFS. In questo modo non perderà tempo a caricare questi jar sui worker.
spark.yarn.archive=hdfs:///tmp/spark-archive.zipPer un funzionamento più veloce, si consiglia di utilizzare kryo come serializzatore. È più ottimizzato rispetto a quello di default.
spark.serializer=org.apache.spark.serializer.KryoSerializerC'è anche un problema noto di Spark, in cui spesso si interrompe a causa della memoria. Questo accade frequentemente quando i lavoratori hanno completato i calcoli e stanno inviando i risultati al driver. Abbiamo aumentato questo parametro. Di default è 1 GB, noi lo abbiamo impostato a 3.
spark.driver.maxResultSize=3072Infine, come dessert. Come aggiornare Spark alla versione 2.1 sulla distribuzione HortonWorks — HDP 2.5.3.0. Questa versione di HDP include una versione preinstallata 2.0, ma un giorno abbiamo deciso che Spark si stava sviluppando rapidamente e che ogni nuova versione risolveva alcuni bug e offriva ulteriori funzionalità, incluse quelle per l'API python, quindi abbiamo deciso di procedere con l'aggiornamento.
Abbiamo scaricato la versione dal sito ufficiale per Hadoop 2.7. L'abbiamo estratta e posizionata nella cartella di HDP. Abbiamo creato i collegamenti simbolici necessari. Lanciamo — ma non si avvia. Mostra un errore piuttosto poco chiaro.
java.lang.NoClassDefFoundError: com/sun/jersey/api/client/config/ClientConfigCercando su Google, abbiamo scoperto che Spark ha deciso di non aspettare che Hadoop si evolvesse e ha scelto di utilizzare una nuova versione di jersey. Lì si stanno già confrontando su questo argomento in JIRA. La soluzione è stata: scaricare . Copiarlo nella cartella jars in SPARK_HOME, ricomprimere in zip e caricarlo su HDFS.
Abbiamo aggirato questo errore, ma ne è emerso uno nuovo e piuttosto complesso.
org.apache.spark.SparkException: L'applicazione Yarn è già terminata! Potrebbe essere stata terminata o non è riuscita a avviare il master dell'applicazioneNel frattempo, stiamo provando a eseguire la versione 2.0 — tutto ok. Prova a indovinare qual è il problema. Siamo entrati nei log di questa applicazione e abbiamo visto qualcosa del genere:
/usr/hdp/${hdp.version}/hadoop/lib/hadoop-lzo-0.6.0.${hdp.version}.jarIn generale, per qualche motivo hdp.version non è stato risolto. Cercando su Google, abbiamo trovato una soluzione. Bisogna entrare in Ambari nelle impostazioni di YARN e aggiungere il parametro personalizzato in yarn-site:
hdp.version=2.5.3.0-37Questa magia ha funzionato e Spark ha decollato. Abbiamo testato alcuni dei nostri notebook jupyter. Tutto funziona. Siamo pronti per il primo corso su Spark che si terrà sabato (già domani)!
UPD. Durante la lezione è emerso un altro problema. A un certo punto, YARN ha smesso di fornire contenitori per Spark. Bisognava correggere un parametro in YARN, che di default era impostato su 0.2:
yarn.scheduler.capacity.maximum-am-resource-percent=0.8 Cioè solo il 20% delle risorse era coinvolto nella distribuzione delle risorse. Modificando i parametri, abbiamo riavviato YARN. Il problema è stato risolto e gli altri partecipanti sono stati in grado di avviare il contesto spark.
Fonte: habr.com
