Ciao Habr! Ieri a , ci sono state molte domande da parte dei partecipanti, relative alla configurazione di questo strumento, da parte dei ragazzi di Rambler&Co. Abbiamo deciso di condividere la nostra esperienza. L'argomento non è semplice - quindi proponiamo di condividere anche la vostra esperienza nei commenti, magari abbiamo anche noi delle incomprensioni.
Una breve introduzione - come utilizziamo Spark. Abbiamo un programma di tre mesi , e durante il secondo modulo i nostri partecipanti lavorano con questo strumento. Di conseguenza, il nostro compito, come organizzatori, è preparare il cluster per l'uso in questo contesto.
La particolarità del nostro utilizzo è che il numero di persone che lavorano contemporaneamente su Spark può essere equivalente all'intero gruppo. Ad esempio, durante il seminario, quando tutti provano qualcosa contemporaneamente e seguono il nostro insegnante. E si tratta di un numero considerevole - a volte quasi 40 persone. Probabilmente non ci sono molte aziende al mondo che affrontano un tale scenario di utilizzo.
Ora vi racconterò come e perché abbiamo scelto determinati parametri di configurazione.
Iniziamo dall'inizio. Spark ha 3 opzioni per lavorare su un cluster: standalone, utilizzando Mesos e utilizzando YARN. Abbiamo deciso di scegliere la terza opzione, perché per noi era logica. Abbiamo già un cluster Hadoop. I nostri partecipanti lo conoscono già bene nella sua architettura. Utilizziamo YARN.
spark.master=yarnOra diventa interessante. Ognuna di queste 3 opzioni di distribuzione ha 2 modalità di deployment: client e cluster. Secondo e vari link su Internet, si può concludere che il client è adatto per il lavoro interattivo - ad esempio, tramite un notebook jupyter, mentre il cluster è più adatto per soluzioni di produzione. Nel nostro caso ci interessava il lavoro interattivo, quindi:
spark.deploy-mode=clientIn generale, da questo momento Spark inizierà a funzionare in qualche modo su YARN, ma non ci bastava. Poiché abbiamo un programma sui Big Data, a volte i partecipanti sentivano la mancanza di ciò che si otteneva con una distribuzione uniforme delle risorse. E qui abbiamo trovato una cosa interessante: l'allocazione dinamica delle risorse. In sintesi, se hai un compito pesante e il cluster è libero (ad esempio, al mattino), utilizzando questa opzione Spark può fornirti risorse aggiuntive. La necessità è calcolata secondo una formula astuta. Non entreremo nei dettagli, funzionano piuttosto bene.
spark.dynamicAllocation.enabled=trueAbbiamo impostato questo parametro, e all'avvio Spark ha segnalato un errore e non si è avviato. Giusto, perché bisognava leggere con più attenzione. È indicato che affinché tutto funzioni, è necessario attivare un parametro aggiuntivo.
spark.shuffle.service.enabled=trueA cosa serve? Quando il nostro job non richiede più così tante risorse, Spark deve restituirle al pool comune. La fase più dispendiosa in termini di tempo in quasi qualsiasi compito MapReduce è la fase di Shuffle. Questo parametro consente di conservare i dati che si generano in questa fase e, di conseguenza, liberare gli executor. Un executor è il processo che calcola tutto sul worker. Ha una certa quantità di core CPU e una certa quantità di memoria.
Abbiamo aggiunto questo parametro. Tutto sembrava funzionare. È diventato evidente che ai partecipanti venivano effettivamente fornite più risorse quando ne avevano bisogno. Ma è emerso un altro problema: a un certo punto, altri partecipanti si svegliavano e volevano anche utilizzare Spark, mentre era tutto occupato, e non erano soddisfatti. Si possono capire. Abbiamo iniziato a consultare la documentazione. Si è scoperto che c'erano anche un certo numero di parametri tramite i quali si poteva influenzare il processo. Ad esempio, se l'executor è in modalità attesa, dopo quanto tempo si possono riprendere le risorse?
spark.dynamicAllocation.executorIdleTimeout=120sNel nostro caso, se i vostri executors non fanno nulla per due minuti, per favore riportateli nel pool generale. Ma nemmeno questo parametro 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 anche un parametro speciale: dopo quanto tempo ritirare gli executors che contengono dati nella cache. Di default, questo parametro era impostato su: infinity! L'abbiamo corretto.
spark.dynamicAllocation.cachedExecutorIdleTimeout=600sCioè, se i vostri executors non fanno nulla per 5 minuti, restituiteli al pool generale. In questo modo, la velocità di rilascio e assegnazione delle risorse per un gran numero di utenti è diventata soddisfacente. Il numero di insoddisfazioni è diminuito. Ma abbiamo deciso di andare oltre e limitare il numero massimo di executors per una singola applicazione — fondamentalmente per un singolo partecipante al programma.
spark.dynamicAllocation.maxExecutors=19Ora, ovviamente, ci sono stati scontenti dall'altra parte — "il cluster è inattivo e ho solo 19 executors", ma cosa possiamo fare — è necessario trovare un giusto equilibrio. Non è possibile rendere tutti felici.
E un'altra piccola storia, legata alla specificità del nostro caso. Un giorno, alcune persone sono arrivate in ritardo a una lezione pratica, e per qualche motivo Spark non è partito. Abbiamo guardato al numero di risorse disponibili — sembrava che ci fossero. Spark dovrebbe partire. Per fortuna, a quel punto la documentazione era già nella nostra mente, e ci siamo ricordati che all'avvio Spark cerca una porta su cui avviarsi. Se la prima porta dell'intervallo è occupata, passa a quella successiva. Se è libera, la occupa. E c'è un parametro che indica il numero massimo di tentativi per questo. Di default, è impostato su 16. Questo numero è inferiore a quello delle persone nel nostro gruppo per la lezione. Di conseguenza, dopo 16 tentativi, Spark abbandonava e diceva che non riusciva a partire. Abbiamo corretto questo parametro.
spark.port.maxRetries=50Ora parlerò di alcune configurazioni, non molto correlate alla specificità del nostro caso.
Per un avvio più rapido di Spark, è consigliato archiviare la cartella jars, che si trova nella directory principale di SPARK_HOME, e caricarla su HDFS. In questo modo non perderà tempo a caricare questi jar sui worker.
spark.yarn.archive=hdfs:///tmp/spark-archive.zipÈ consigliabile utilizzare kryo come serializzatore per un funzionamento più veloce. È più ottimizzato rispetto a quello predefinito.
spark.serializer=org.apache.spark.serializer.KryoSerializerC'è anche un problema noto con Spark, che spesso va in crash per mancanza di memoria. Questo accade frequentemente quando i worker hanno completato i calcoli e inviano il risultato al driver. Abbiamo impostato questo parametro su un valore maggiore. Di default è 1GB, noi lo abbiamo impostato a 3.
spark.driver.maxResultSize=3072E infine, 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 sviluppa molto rapidamente, e ogni nuova versione risolve alcuni bug e offre ulteriori funzionalità, incluso il supporto per l'API Python, quindi abbiamo deciso di fare un 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. Tentiamo di avviare — ma non si avvia. Riporta un errore 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 usare una nuova versione di jersey. Si stanno confrontando su questo tema nella JIRA. La soluzione è stata — scaricare . Metterlo nella cartella jars in SPARK_HOME, ricreare il zip e caricarlo su HDFS.
Abbiamo bypassato questo errore, ma ne è emerso uno nuovo e piuttosto evasivo.
org.apache.spark.SparkException: L'applicazione Yarn è già terminata! Potrebbe essere stata interrotta o non riuscita a avviare l'application masterNel frattempo proviamo a eseguire la versione 2.0 — tutto ok. Prova a indovinare quale sia 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}.jarDunque, per qualche motivo hdp.version non era risolvibile. Cercando online, abbiamo trovato la soluzione. È necessario accedere alle impostazioni di YARN in Ambari e aggiungere il parametro nelle impostazioni personalizzate di yarn-site:
hdp.version=2.5.3.0-37Questa magia ha funzionato e Spark ha iniziato a funzionare. Abbiamo testato alcuni dei nostri notebook Jupyter. Tutto funziona. Siamo pronti per la prima lezione di Spark di sabato (già domani)!
UPD. Durante la lezione è emerso un altro problema. A un certo punto YARN ha smesso di fornire contenitori per Spark. Era necessario modificare un parametro in YARN che di default era impostato a 0.2:
yarn.scheduler.capacity.maximum-am-resource-percent=0.8 Cioè solo il 20% delle risorse ha partecipato alla distribuzione delle risorse. Modificando i parametri, abbiamo riavviato YARN. Il problema è stato risolto e anche gli altri partecipanti sono riusciti a avviare il contesto spark.
Fonte: habr.com
