Définissez la taille de la mémoire du pilote de manière programmatique dans PySpark

Question

Pour écrire un script autonome, je voudrais démarrer et configurer un contexte Spark directement depuis Python. En utilisant le script PySpark, je peux définir la taille de mémoire du driver avec :

$ /opt/spark-1.6.1/bin/pyspark
... INFO MemoryStore: MemoryStore started with capacity 511.5 MB ...
$ /opt/spark-1.6.1/bin/pyspark --conf spark.driver.memory=10g
... INFO MemoryStore: MemoryStore started with capacity 7.0 GB ...

Mais lorsque je démarre le contexte depuis le module Python, la taille de la mémoire du driver ne peut pas être définie :

$ export SPARK_HOME=/opt/spark-1.6.1                                                                                                                                                                                                                                                                                                                
$ export PYTHONPATH=$PYTHONPATH:$SPARK_HOME/python
$ python
>>> from pyspark import SparkConf, SparkContext
>>> sc = SparkContext(conf=SparkConf().set('spark.driver.memory', '10g'))
... INFO MemoryStore: MemoryStore started with capacity 511.5 MB ...

La seule solution que je connaisse est de définir spark.driver.memory dans sparks-default.conf, ce qui n'est pas satisfaisant. Comme expliqué dans ce post, il est logique pour Java/Scala de ne pas pouvoir changer la taille de mémoire du driver une fois que le JVM est démarré. Y a-t-il un moyen de le configurer dynamiquement depuis Python avant ou lors de l'importation du module pyspark ?

Answer 1

Il est inutile d'utiliser la conf comme vous le faites. Essayez d'ajouter ce préambule à votre code :

memory = '10g'
pyspark_submit_args = ' --driver-memory ' + memory + ' pyspark-shell'
os.environ["PYSPARK_SUBMIT_ARGS"] = pyspark_submit_args

Answer 2

J'avais exactement le même problème et je viens de découvrir un moyen astucieux de le résoudre. Et il s'avère qu'il existe une réponse existante qui adopte la même approche. Mais je vais expliquer pourquoi cela fonctionne.

Comme vous le savez, la mémoire du pilote ne peut pas être définie après que le JVM démarre. Mais lors de la création d'un SparkContext, pyspark démarre le JVM en appelant spark-submit et en passant pyspark-shell en tant que commande

SPARK_HOME = os.environ["SPARK_HOME"]
# Lancer la passerelle Py4j en utilisant la commande run de Spark afin de récupérer le
# classpath et les paramètres corrects de spark-env.sh
on_windows = platform.system() == "Windows"
script = "./bin/spark-submit.cmd" if on_windows else "./bin/spark-submit"
submit_args = os.environ.get("PYSPARK_SUBMIT_ARGS", "pyspark-shell")
if os.environ.get("SPARK_TESTING"):
   submit_args = ' '.join([
        "--conf spark.ui.enabled=false",
        submit_args
    ])
command = [os.path.join(SPARK_HOME, script)] + shlex.split(submit_args)

Remarquez la variable d'environnement PYSPARK_SUBMIT_ARGS. Ce sont les arguments que le contexte enverra à la commande spark-submit.

Donc tant que vous définissez PYSPARK_SUBMIT_ARGS="--driver-memory=2g pyspark-shell" avant d'instancier un nouveau SparkContext, le paramètre de mémoire du pilote devrait prendre effet. Il existe plusieurs façons de définir cette variable d'environnement, consultez la réponse que j'ai liée précédemment pour une méthode.

Answer 3

Vous pouvez le passer à travers la commande spark-submit en utilisant le drapeau --driver-memory.

spark-submit   \
    --master yarn  \
    --deploy-mode cluster  \
    --driver-cores 12 --driver-memory 20g \
    --num-executors 52 --executor-cores 6  --executor-memory 30g MySparkApp.py

Placez cette commande ci-dessus dans un script shell ou autre et au lieu de 20 (la mémoire des pilotes qui est définie manuellement) utilisez une variable que vous pouvez changer dynamiquement.