Lindorm:Développement de jobs Python : guide pratique

Prérequis

Avant de commencer, assurez-vous de disposer des éléments suivants :

• Un moteur de calcul Lindorm activé. Voir Activation du service

• Un bucket OSS pour stocker les fichiers de votre projet Python, l'environnement d'exécution et le script de lancement

• Un AccessKey ID et un AccessKey Secret disposant des autorisations de lecture et d'écriture sur le bucket OSS. Voir Création d'une paire de clés d'accès

• Un environnement Linux (nécessaire pour l'empaquetage des fichiers binaires Python compatibles avec le moteur de calcul Lindorm)

Fonctionnement

Le flux de soumission des jobs Spark Python comporte quatre étapes :

1. Définir le job — structurez votre projet Python avec les fichiers de point d'entrée requis.

2. Empaqueter le job — regroupez séparément le code de votre projet et l'environnement d'exécution Python.

3. Charger les fichiers — stockez tous les artefacts dans OSS.

4. Soumettre le job — exécutez le job depuis la console Lindorm ou Data Management Service (DMS).

Étape 1 : Définir le job

Téléchargez le package d'exemple de job Spark et extrayez-le. Le dossier extrait est nommé lindorm-spark-examples. Consultez le répertoire lindorm-spark-examples/python pour obtenir un exemple de structure de projet.

La racine de votre projet (your_project dans l'exemple) nécessite trois modifications structurelles avant sa soumission.

1. Ajouter __init__.py

Créez un fichier vide __init__.py dans le répertoire your_project. Cette opération transforme le répertoire en module Python importable par launcher.py.

2. Préparer main.py

Ouvrez your_project/main.py et apportez les deux modifications suivantes :

Ajoutez le répertoire du projet à sys.path afin que les imports soient résolus correctement au moment de l'exécution :

current_dir = os.path.abspath(os.path.dirname(__file__))
sys.path.append(current_dir)

Enveloppez votre logique d'entrée dans une fonction main(argv) :

def main(argv):
    # Write your job logic here
    pass

if __name__ == "__main__":
    main(sys.argv)

Pour un exemple fonctionnel, le code suivant initialise une SparkSession :

from pyspark.sql import SparkSession

def main(argv):
    spark = SparkSession \
        .builder \
        .appName("PythonImportTest") \
        .getOrCreate()
    print(spark.conf)
    spark.stop()

if __name__ == "__main__":
    main(sys.argv)

3. Créer launcher.py

Dans le répertoire racine your_project, créez un fichier nommé launcher.py. Copiez le contenu depuis lindorm-spark-examples/python/launcher.py. Ce fichier constitue le point d'entrée appelé par le moteur de calcul Lindorm ; il ajoute le répertoire du projet à sys.path et appelle votre fonction main(argv).

Étape 2 : Empaqueter le job

L'empaquetage produit deux artefacts distincts : un fichier .zip (ou .egg) contenant le code de votre projet, et un fichier tar contenant l'environnement d'exécution Python.

zip -r your_project.zip your_project

conda create -y -n pyspark_conda_env -c conda-forge numpy conda-pack
conda activate pyspark_conda_env
conda pack -f -o pyspark_conda_env.tar.gz

Étape 3 : Charger les fichiers dans OSS

Chargez les trois artefacts dans votre bucket OSS. Voir Chargement simple.

• launcher.py — le point d'entrée créé à l'étape 1

• your_project.zip (ou .egg) — le package de code du projet issu de l'étape 2

• pyspark_conda_env.tar.gz — l'environnement d'exécution Python issu de l'étape 2

Étape 4 : Soumettre le job

Le moteur de calcul Lindorm prend en charge deux méthodes de soumission :

• Console Lindorm — Voir Gestion des jobs dans la console.

• DMS — Voir Gestion des jobs avec DMS.

Quelle que soit la méthode, configurez les paramètres suivants dans le champ configs de votre demande de soumission de job.

{
  "spark.archives": "oss://testBucketName/pyspark_conda_env.tar.gz#environment",
  "spark.kubernetes.driverEnv.PYSPARK_PYTHON": "./environment/bin/python",
  "spark.submit.pyFiles": "oss://testBucketName/your_project.zip",
  "spark.hadoop.fs.oss.endpoint": "oss-cn-beijing-internal.aliyuncs.com",
  "spark.hadoop.fs.oss.accessKeyId": "<your-access-key-id>",
  "spark.hadoop.fs.oss.accessKeySecret": "<your-access-key-secret>",
  "spark.hadoop.fs.oss.impl": "org.apache.hadoop.fs.aliyun.oss.AliyunOSSFileSystem"
}

Diagnostic des jobs

Après avoir soumis le job, consultez son statut et l'adresse de l'interface utilisateur Spark sur la page Jobs. Voir Affichage d'un job.

En cas d'échec de la soumission, ouvrez un ticket et fournissez l'ID du job ainsi que l'adresse de l'interface utilisateur Spark à l'équipe d'assistance.

Étapes suivantes

• Gestion des jobs dans la console

• Gestion des jobs avec DMS