Spark avec SSH sur Qarnot Cloud

Spark est un moteur rapide et général pour le traitement et le calcul de données à grande échelle sur un cluster distribué. Apache Spark offre un mode de déploiement simple et autonome qui utilise son propre gestionnaire de ressources et permet la création d'une architecture distribuée maître-esclave (master-slave).

Voici un guide rapide étape par étape sur la manière de configurer un Cluster Spark Standalone sur Qarnot, de s'y connecter via un tunneling SSH et de soumettre une application Spark qui compte le nombre de mots dans l'Iliade.

Version

• Hadoop : 3.2 sorti en 2021
• Spark : 3.1.1 sorti en 2021

Si vous êtes intéressé par une autre version, veuillez nous envoyer un e-mail à qlab@qarnot.com.

Prérequis

Avant de démarrer un calcul avec le SDK Python, quelques étapes sont nécessaires :

• Récupérer le jeton d'authentification
• Installer le SDK Python de Qarnot (ici)

Note : En plus du SDK Python, Qarnot fournit des SDK C# et Node.js ainsi qu'une ligne de commande.

Cas test

Ce tutoriel montrera comment démarrer un cluster Spark Qarnot depuis votre ordinateur en suivant ces étapes simples :

• Démarrer le cluster sur Qarnot
• S'y connecter via le tunneling SSH
• Soumettre votre application via l'interface en ligne de commande (CLI)

Avant de continuer, vous devriez configurer votre environnement de travail pour qu'il contienne les fichiers suivants, téléchargeables ici :

• input :
  
  • ‍iliad.txt : fichier texte contenant l'Iliade dont les mots doivent être comptés sur Qarnot.
  • word_count.py : script pour compter le nombre de mots dans iliad.txt.
• ssh-spark.py : script pour démarrer le cluster sur Qarnot, voir ci-dessous.

Lancement du cas test

Une fois que tout est configuré, utilisez le script suivant dans votre terminal pour démarrer le cluster sur Qarnot.

Assurez-vous de copier votre jeton d'authentification et votre clé ssh publique (à la place de <<<MY_SECRET_TOKEN>>> et <<<PUBLIC_SSH_KEY>>>) dans le script ssh-spark.py. Par défaut, votre clé SSH publique se trouve dans ~/.ssh/<<<nom_cle_ssh>>>.pub.

• Assurez-vous de copier votre jeton d'authentification à la place de <<<MY_SECRET_TOKEN>>> à la ligne 10.
• Copiez votre clé ssh publique dans le script à la place de <<<PUBLIC_SSH_KEY>>> à la ligne 26.
• Enfin, vous pouvez changer <<<PORT>>> à la ligne 66 pour le port que vous souhaitez utiliser pour le tunneling SSH avec Qarnot.

ssh-spark.py

#!/usr/bin/env python3

# Import the Qarnot SDK
import sys
import argparse
import qarnot
import subprocess

# Connect to the Qarnot platform
conn = qarnot.Connection(client_token = '<<<MY_SECRET_TOKEN>>>')

# Create a task
task = conn.create_task('Hello World - SSH-Spark', 'docker-cluster-network-ssh', 3)

# Create a resource bucket and add input files
input_bucket = conn.create_bucket('ssh-spark-in')
input_bucket.sync_directory('input/')

# Attach the bucket to the task
task.resources.append(input_bucket)

# Create a result bucket and attach it to the task
task.results = conn.create_bucket('ssh-spark-out')

# Task constants are the main way of controlling a task's behaviour
task.constants['DOCKER_SSH'] = '<<<PUBLIC_SSH_KEY>>>'
task.constants['DOCKER_REPO'] = "qarnotlab/spark"
task.constants['DOCKER_TAG'] = "v3.1.2"
task.constants['DOCKER_CMD_MASTER'] = "/bin/bash \
                                      /opt/start-master.sh ${INSTANCE_COUNT} /opt/log.sh"
task.constants['DOCKER_CMD_WORKER'] = "/bin/bash -c '/opt/start-worker.sh /opt/log.sh && sleep infinity'"

# Submit the task to the Api, that will launch it on the cluster
task.submit()
error_happened = False

# Initial values for forward port and ssh tunneling bool
ssh_forward_port = None
ssh_tunneling_done = False

# Get optional terminal app name from argument
parser = argparse.ArgumentParser()
parser.add_argument("--terminal", type=str,
                    help="Unix terminal app to be used for ssh connexion", 
                    default = 'gnome-terminal', required=False)
args = parser.parse_args()

# If not provided by the user it will be set to gnome-terminal by default
terminal = args.terminal
done = False
while not done:

# Wait for the task to be FullyExecuting
    if task.state == 'FullyExecuting' :
    
        # If the ssh connexion was not done yet and the list active_forward is available (len!=0)
        active_forward = task.status.running_instances_info.per_running_instance_info[0].active_forward
        if not ssh_tunneling_done and len(active_forward)!=0:
            ssh_forward_port = active_forward[0].forwarder_port
            
            # Once the port has been fetched, spawn a new terminal with the ssh command
            cmd = 'ssh -L <<<PORT>>>:localhost:6060 -o StrictHostKeyChecking=no root@forward01.qarnot.net -p '+ str(ssh_forward_port)
            
            # if --terminal was set to off, bypass this and connect manually with ssh
            if terminal != 'off':
                ssh_cmd = [terminal,'-e',cmd]
                subprocess.call(ssh_cmd)
            else:
                print("\n** Run this command in your terminal to connect via ssh **\n", cmd,
                        "\n**********************************************************")
                        
            # set this var to True in order to not run ssh again
            ssh_tunneling_done = True
            
    # sync ouptput files of the qarnot machine with the bucket,
    # then the bucket with your local directory
    task.instant()
    done = task.wait(5)
    task.download_results('outputs')

Pour lancer ce script, copiez simplement le code ci-dessus dans un script Python et exécutez python3 ssh-spark.py dans votre terminal.

Par défaut, il vous connectera à Qarnot via ssh dans un gnome-terminal. Si vous n'avez pas cette application de terminal installée ou si vous souhaitez en utiliser une autre, vous pouvez exécuter python3 run.py --terminal=<<<application-terminal-unix>>>. De plus, si vous souhaitez désactiver cette fonctionnalité et imprimer uniquement la commande que vous pouvez exécuter vous-même dans votre terminal, vous pouvez définir --terminal=off.

Une fois qu'un nouveau terminal apparaît sur votre écran, cela signifie que la connexion ssh avec le cluster est sécurisée. Tout ce que vous avez à faire est d'exécuter les commandes suivantes dans votre terminal ssh :

	# commands to source the environment variables in the shell
PID=$(cat /tmp/pid_start_master)
. <(xargs -0 bash -c 'printf "export %q\n" "$@"' -- < /proc/$PID/environ)

# submit an app to the cluster to count the number of words in the Iliad
/spark/bin/spark-submit --master $SPARK_MASTER --driver-memory 2G --executor-memory 14G /job/word_count.py `sed -n 1p /share/qhost`

Si le terminal apparaît et disparaît rapidement, cela signifie très probablement que le port que vous avez choisi est actuellement occupé et que la connexion ne peut pas être établie. Vous pouvez changer de port et réessayer.

Il est possible d'accéder au tableau de bord Spark UI transféré via tunneling SSH en tapant localhost:<<<PORT>>> dans votre navigateur. Vous pouvez surveiller l'état de votre cluster et les différentes applications que vous avez soumises.

Résultats

À tout moment, vous pouvez surveiller l'état de votre tâche sur notre plateforme.

Le résultat du comptage est écrit dans un fichier texte output_iliad.txt que l'on peut trouver dans votre output bucket spark-ssh-out, ainsi que les logs du maître (master) et des travailleurs (worker), qui peuvent tous être consultés dans votre output bucket.

Une fois que vous avez terminé avec la tâche, tapez simplement exit dans le terminal ssh pour fermer le tunneling et assurez-vous d'interrompre la tâche depuis notre plateforme. Si vous n'interrompez pas la tâche manuellement, elle continuera de s'exécuter et utilisera vos crédits.

Conclusion

C'est tout ! Si vous avez des questions, veuillez contacter qlab@qarnot.com et nous vous aiderons avec plaisir !