Configuration de Spark sur YARN

Habr, bonjour ! Hier, lors de la réunion dédiée à Apache Spark, des questions ont été posées par de nombreux participants concernant la configuration de cet outil. Nous avons décidé de partager notre expérience suite à cela. Le sujet n'est pas simple — nous vous invitons donc à également partager votre expérience dans les commentaires, peut-être que nous avons compris ou utilisé quelque chose de manière incorrecte.

Une petite introduction — comment nous utilisons Spark. Nous avons un programme de trois mois « Spécialiste des grandes données », et durant tout le deuxième module, nos participants travaillent avec cet outil. Par conséquent, notre tâche en tant qu'organisateurs est de préparer un cluster pour une utilisation dans ce cadre.

Une particularité de notre utilisation est que le nombre de personnes travaillant simultanément sur Spark peut être égal à l'ensemble du groupe. Par exemple, lors d'un séminaire, lorsque tout le monde essaie et répète ce que notre enseignant propose. Cela peut parfois aller jusqu'à 40 personnes. Il n'y a probablement pas beaucoup d'entreprises dans le monde qui rencontrent un tel scénario d'utilisation.

Je vais ensuite expliquer comment et pourquoi nous avons choisi certains paramètres de configuration.

Commençons par le début. Spark peut fonctionner sur le cluster de 3 manières : en mode standalone, en utilisant Mesos ou en utilisant YARN. Nous avons choisi la troisième option car elle nous semblait logique. Nous avons déjà un cluster hadoop. Nos participants connaissent bien son architecture. Utilisons YARN.

spark.master=yarn

Ensuite, ça devient plus intéressant. Chacune de ces 3 options de déploiement a 2 modes de déploiement : client et cluster. D'après documentation et divers liens sur Internet, on peut conclure que le mode client convient pour un travail interactif — par exemple, via un notebook jupyter, tandis que le mode cluster est plus adapté aux solutions de production. Dans notre cas, nous étions intéressés par un travail interactif, donc :

spark.deploy-mode=client

En gros, à partir de ce moment là, Spark fonctionnera déjà d'une certaine manière sur YARN, mais cela ne nous suffisait pas. Comme notre programme porte sur les Big Data, il arrivait parfois que les participants manquent de ce qui se présentait dans le cadre d'une répartition uniforme des ressources. Et là, nous avons découvert une chose intéressante : l'allocation dynamique des ressources. Pour faire court, l'idée est la suivante : si vous avez une tâche lourde et que le cluster est libre (par exemple, le matin), grâce à cette option, Spark peut vous fournir des ressources supplémentaires. Le besoin est calculé selon une formule astucieuse. Nous ne rentrerons pas dans les détails — cela fonctionne assez bien.

spark.dynamicAllocation.enabled=true

Nous avons défini ce paramètre, et lors du lancement de Spark, une erreur s'est produite et il ne s'est pas lancé. C'est normal, car il fallait lire documentation plus attentivement. Il est indiqué que pour que tout soit en ordre, il faut également activer un paramètre supplémentaire.

spark.shuffle.service.enabled=true

Quel est son besoin ? Lorsque notre job ne nécessite plus autant de ressources, Spark doit les renvoyer au pool général. La phase la plus coûteuse en main-d'œuvre dans presque n'importe quelle tâche MapReduce est la phase de Shuffle. Ce paramètre permet de conserver les données générées durant cette phase et ainsi de libérer les executors. Un executor est un processus qui calcule tout sur le worker. Il dispose d'un certain nombre de cœurs processeurs et d'une certaine quantité de mémoire.

Nous avons ajouté ce paramètre. Tout semblait fonctionner. Il est devenu évident que les participants recevaient réellement plus de ressources quand ils en avaient besoin. Mais un autre problème est survenu : à un moment donné, d'autres participants se réveillaient et voulaient aussi utiliser Spark, mais tout était occupé, et ils étaient mécontents. On peut les comprendre. Nous avons commencé à consulter la documentation. Il s'est avéré qu'il existe encore un certain nombre de paramètres qui peuvent influencer le processus. Par exemple, si un executor est en mode attente — au bout de combien de temps peut-on récupérer les ressources ?

spark.dynamicAllocation.executorIdleTimeout=120s

Dans notre cas, si vos executors ne font rien pendant deux minutes, veuillez les renvoyer dans le pool général. Cependant, ce paramètre n'était pas toujours suffisant. Il était évident qu'une personne ne faisait rien depuis longtemps, mais les ressources n'étaient pas libérées. Il s'est avéré qu'il existe un paramètre spécial pour déterminer au bout de combien de temps récupérer les executors contenant des données en cache. Par défaut, ce paramètre était défini sur — infini ! Nous l'avons corrigé.

spark.dynamicAllocation.cachedExecutorIdleTimeout=600s

Ainsi, si vos executors ne font rien pendant 5 minutes, renvoyez-les dans le pool général. Dans ce mode, la vitesse de libération et de distribution des ressources pour un grand nombre d'utilisateurs est devenue satisfaisante. Le nombre de mécontents a diminué. Mais nous avons décidé d'aller plus loin et de limiter le nombre maximum d'executors pour une application — en gros, pour un participant au programme.

spark.dynamicAllocation.maxExecutors=19

Bien sûr, il y a maintenant des mécontents de l'autre côté — "le cluster est inactif, et je n'ai que 19 executors", mais que peut-on faire ? Il faut un certain équilibre correct. On ne peut pas rendre tout le monde heureux.

Et une autre petite histoire liée à la spécificité de notre cas. Un jour, plusieurs personnes sont arrivées en retard à la séance pratique, et leur Spark ne s'est pas lancé pour une raison quelconque. Nous avons vérifié le nombre de ressources disponibles — apparemment, il y en avait. Spark devait se lancer. Heureusement, d'ici là, la documentation était déjà ancrée dans notre mémoire, et nous nous sommes souvenus que lors du lancement, Spark recherche un port sur lequel se lancer. Si le premier port de la plage est occupé, il passe au suivant. S'il est libre, il l'attrape. Et il existe un paramètre qui indique le nombre maximum de tentatives à cet effet. Par défaut, il est de 16. Ce nombre est inférieur au nombre de personnes dans notre groupe de séance. Donc, après 16 tentatives, Spark abandonnait et disait qu'il ne pouvait pas démarrer. Nous avons corrigé ce paramètre.

spark.port.maxRetries=50

Je vais maintenant parler de certains réglages qui ne sont pas vraiment liés à la spécificité de notre cas.

Pour un démarrage plus rapide de Spark, il est recommandé de compresser le dossier jars situé dans le répertoire domestique de SPARK_HOME et de le placer sur HDFS. Ainsi, il ne perdra pas de temps à charger ces jar files sur les workers.

spark.yarn.archive=hdfs:///tmp/spark-archive.zip

Il est également recommandé d'utiliser kryo comme sérialiseur pour un fonctionnement plus rapide. Il est plus optimisé que celui par défaut.

spark.serializer=org.apache.spark.serializer.KryoSerializer

Il y a un vieux problème avec Spark, c'est qu'il tombe souvent à court de mémoire. Cela se produit souvent lorsque les workers ont tout calculé et envoient le résultat au driver. Nous avons augmenté ce paramètre. Par défaut, il est de 1 Go, nous l'avons porté à 3.

spark.driver.maxResultSize=3072

Et enfin, comme dessert. Comment mettre à jour Spark vers la version 2.1 sur la distribution HortonWorks - HDP 2.5.3.0. Cette version de HDP contient une version préinstallée 2.0, mais un jour, nous avons décidé que Spark évoluait activement et que chaque nouvelle version corrigeait certains bugs tout en offrant des fonctionnalités supplémentaires, y compris pour l'API Python, donc nous avons décidé de faire la mise à jour.

Nous avons téléchargé la version sur le site officiel pour Hadoop 2.7. Nous l'avons décompressée, placée dans le dossier HDP. Nous avons créé les liens symboliques nécessaires. Nous lançons - ça ne démarre pas. Un message d'erreur très incompréhensible s'affiche.

java.lang.NoClassDefFoundError: com/sun/jersey/api/client/config/ClientConfig

En cherchant sur Google, nous avons découvert que Spark a décidé de ne pas attendre que Hadoop se développe, et a choisi d'utiliser une nouvelle version de jersey. Ils se disputent à ce sujet sur JIRA. La solution était de télécharger jersey version 1.17.1. Nous l'avons placé dans le dossier jars de SPARK_HOME, à nouveau zipé et transféré sur HDFS.

Nous avons contourné cette erreur, mais une nouvelle erreur plutôt obscure est apparue.

org.apache.spark.SparkException: L'application Yarn est déjà terminée ! Elle a peut-être été tuée ou n'a pas pu lancer le maître d'application.

En essayant de lancer la version 2.0 — tout va bien. Essaie de deviner quel est le problème. Nous avons plongé dans les logs de cette application et avons vu quelque chose comme ça :

/usr/hdp/${hdp.version}/hadoop/lib/hadoop-lzo-0.6.0.${hdp.version}.jar

En gros, pour une raison quelconque, hdp.version n'a pas été résolue. Après avoir cherché sur Google, nous avons trouvé une solution. Il faut aller dans Ambari, dans les paramètres YARN, et ajouter ce paramètre dans yarn-site personnalisé :

hdp.version=2.5.3.0-37

Cette magie a fonctionné, et Spark a redémarré. Nous avons testé quelques-uns de nos jupyter-notebooks. Tout fonctionne. Nous sommes prêts pour notre premier cours sur Spark ce samedi (déjà demain) !

MAJ. Au cours, nous avons découvert un autre problème. À un moment donné, YARN a cessé de fournir des conteneurs pour Spark. Il fallait corriger un paramètre dans YARN, qui par défaut était à 0.2 :

yarn.scheduler.capacity.maximum-am-resource-percent=0.8

C'est-à-dire que seulement 20 % des ressources ont participé à la distribution des ressources. En modifiant les paramètres, nous avons redémarré YARN. Le problème a été résolu et les autres participants ont également pu lancer le contexte Spark.

Source : habr.com