Soumettre une tâche Spark - E-MapReduce - Centre de documentation Alibaba Cloud

EMR on ACK prend en charge trois méthodes pour soumettre des tâches : l'utilisation d'une définition de ressource personnalisée (CRD), la commande spark-submit ou le terminal de la console. Cette rubrique décrit ces méthodes.

Prérequis

Créez un cluster Spark dans la console EMR on ACK. Pour plus d'informations, consultez la section Créer un cluster.

Notes

Dans les exemples de cette rubrique, le fichier JAR est directement intégré à l'image. Si vous utilisez votre propre fichier JAR, vous pouvez le télécharger vers Object Storage Service (OSS). Pour plus d'informations, consultez la section Téléchargement simple.

Remplacez local:///opt/spark/examples/spark-examples.jar dans la commande par le chemin OSS de votre fichier JAR. Le format du chemin est oss://<yourBucketName>/<path>.jar.

Soumettre une tâche

Méthode 1 : Soumission via CRD

Utilisez kubectl pour vous connecter au cluster Kubernetes. Pour plus d'informations, consultez la section Obtenir le fichier KubeConfig d'un cluster et utiliser kubectl pour se connecter au cluster.

Créez un fichier nommé spark-pi.yaml avec le contenu suivant.

apiVersion: "sparkoperator.k8s.io/v1beta2"
kind: SparkApplication
metadata:
  name: spark-pi-simple
spec:
  type: Scala
  sparkVersion: 3.2.1
  mainClass: org.apache.spark.examples.SparkPi
  mainApplicationFile: "local:///opt/spark/examples/spark-examples.jar"
  arguments:
    - "1000"
  driver:
    cores: 1
    coreLimit: 1000m
    memory: 4g
  executor:
    cores: 1
    coreLimit: 1000m
    memory: 8g
    memoryOverhead: 1g
    instances: 1

Pour obtenir la description des paramètres de cet exemple, consultez la documentation spark-on-k8s-operator.

Remarque

Vous pouvez utiliser un nom de fichier personnalisé. Cette rubrique utilise spark-pi.yaml à titre d'exemple.
L'exemple utilise Spark 3.2.1 (EMR-5.6.0). Si vous utilisez une version différente, modifiez le paramètre sparkVersion.

Exécutez la commande suivante pour soumettre la tâche.
```
kubectl apply -f spark-pi.yaml --namespace <cluster-namespace>
```
Remplacez <cluster-namespace> par le namespace de votre cluster. Vous trouverez le namespace sur la page Cluster Details dans la console EMR on ACK.
Le résultat suivant s'affiche :
```
sparkapplication.sparkoperator.k8s.io/spark-pi-simple created
```
Remarque
spark-pi-simple correspond au nom de la tâche dans cet exemple.
Facultatif : Consultez la tâche créée sur la page Job Details.

Méthode 2 : Commande spark-submit

Utilisez kubectl pour vous connecter au cluster Kubernetes. Pour plus d'informations, consultez la section Obtenir le fichier KubeConfig d'un cluster et utiliser kubectl pour se connecter au cluster.
Exécutez les commandes suivantes pour installer l'outil emr-spark-ack d'Alibaba Cloud E-MapReduce et définir ses permissions.
```
wget https://ecm-repo-cn-hangzhou.oss-cn-hangzhou.aliyuncs.com/emr-on-ack/util/emr-spark-ack
chmod 755 emr-spark-ack
```

Utilisez l'outil emr-spark-ack pour soumettre la tâche.

Utilisez la syntaxe suivante.

 ./emr-spark-ack -n <cluster-namespace> <spark-command>

Remarque

L'espace réservé <spark-command> prend en charge spark-submit, spark-sql, spark-shell et pyspark. La syntaxe est identique à celle des commandes Spark natives.

Exemple : mode cluster

Utilisez spark-submit pour soumettre une tâche spark-pi.

./emr-spark-ack -n <cluster-namespace> spark-submit \
    --name spark-pi-submit \
    --deploy-mode cluster \
    --class org.apache.spark.examples.SparkPi \
    local:///opt/spark/examples/spark-examples.jar \
    1000

Exemple : mode client

Utilisation de la commande spark-sql

# Prepare a local SQL file.
echo "select 1+1">test.sql
# Submit the job.
./emr-spark-ack -n <cluster-namespace> spark-sql -f test.sql

Pour les clusters exécutant Spark 3 ou une version ultérieure (EMR-5.X), l'outil emr-spark-ack télécharge automatiquement les dépendances locales. L'outil transfère les fichiers locaux spécifiés par des paramètres tels que --jars, --files et -f vers le cluster EMR on ACK avant de soumettre la tâche dans l'environnement Kubernetes.

Le code suivant fournit un exemple et affiche le résultat :

[user@hostname ~]$ echo "select 1+1">test.sql
[user@hostname ~]$ ./emr-spark-ack -n <cluster-namespace> spark-sql -f test.sql
uploading test.sql to c-xxx/spark-submit-gateway-865bbf56f6-sfb95:/opt/spark/work-dir/test.sql
cmd: kubectl exec spark-submit-gateway-865bbf56f6-sfb95 -i -n <cluster-namespace> -- /bin/bash spark-sql -f test.sql
Warning: Ignoring non-Spark config property: ack.clusterid.for.rss.linked
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
Spark master: k8s://https://kubernetes.default:443, Application Id: spark-application-1653639654376
Time taken: 4.738 seconds, Fetched 1 row(s)
2
22/05/27 16:21:04 WARN [shutdown-hook-0] ExecutorPodsWatchSnapshotSource: Kubernetes client has been closed.

Utilisation de la commande spark-shell

./emr-spark-ack -n <cluster-namespace> spark-shell

Le code suivant fournit un exemple et affiche le résultat :

[user@hostname ~]$ ./emr-spark-ack -n <cluster-namespace> spark-shell
cmd: kubectl exec spark-submit-gateway-865bbf56f6-sfb95 -it -n <cluster-namespace> -- /bin/bash spark-shell
Warning: Ignoring non-Spark config property: ack.clusterid.for.rss.linked
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
Spark context Web UI available at http://10.210.xxx.xxx:4041
Spark context available as 'sc' (master = k8s://https://kubernetes.default:443, app id = spark-application-16536387xxx).
Spark session available as 'spark'.
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 3.2.1
      /_/
Using Scala version 2.12.15 (OpenJDK 64-Bit Server VM, Java 1.8.0_332)
Type in expressions to have them evaluated.
Type :help for more information.
scala>

Facultatif : Consultez la tâche créée sur la page Job Details.
Facultatif : Utilisez l'outil emr-spark-ack pour arrêter une tâche.
Utilisez la syntaxe suivante.
```
 ./emr-spark-ack -n <cluster-namespace> kill <Spark_app_id>
```
Remarque
L'outil emr-spark-ack génère l'identifiant <Spark_app_id> lors de la soumission de la tâche. Cet ID est disponible dans le journal de sortie.

Méthode 3 : Terminal de la console

Accédez à la page Access Links and Ports.
1. Connectez-vous à la console EMR on ACK.
2. Sur la page EMR on ACK, cliquez sur le nom du cluster cible.
3. Cliquez sur l'onglet Access Links and Ports.
Sur la page Access Links and Ports, cliquez sur le lien SparkSubmitGateway UI.

Le terminal shell s'ouvre.

Dans le terminal shell, exécutez les commandes Spark de l'une des deux manières suivantes.

Utilisation de la commande spark-sql

spark-sql

Une fois dans le shell spark-sql, vous pouvez exécuter des commandes Spark pour des requêtes interactives.

[root@spark-submit-gateway-865bb... work-dir]# spark-sql
Warning: Ignoring non-Spark config property: ack.clusterid.for.rss.linked
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
Spark master: k8s://https://kubernetes.default:443, Application Id: spark-application-1654656192425
spark-sql> select 1+1;
2
Time taken: 5.277 seconds, Fetched 1 row(s)
spark-sql>

Utilisation de la commande spark-submit

spark-submit \
    --name spark-pi-submit \
    --deploy-mode cluster \
    --class org.apache.spark.examples.SparkPi \
    local:///opt/spark/examples/spark-examples.jar \
    1000

Facultatif : Consultez la tâche créée sur la page Job Details.

Rubriques connexes

Pour gérer les tâches Spark avec kubectl, consultez la section Utiliser kubectl pour gérer les tâches.
Pour collecter les journaux des tâches Spark avec Simple Log Service, consultez la section Utiliser Simple Log Service pour collecter les journaux des tâches Spark.
Pour configurer les métadonnées d'un cluster Spark dans EMR on ACK, consultez la section Configurer les métadonnées d'un cluster Spark.
Pour planifier de manière élastique les tâches Spark avec Elastic Container Instance (ECI), consultez la section Utiliser ECI pour planifier de manière élastique les tâches Spark.