Plusieurs travailleurs Python (ou threads de travailleurs) dans PySpark ?

Question

Dans PySpark, je comprends que les travailleurs python sont utilisés pour effectuer (au moins une partie) du calcul sur les nœuds de travail (comme décrit à https://cwiki.apache.org/confluence/display/SPARK/PySpark+Internals).

Dans ma configuration de test, j'essaie de faire en sorte que Spark utilise 4 threads de travailleurs (sur une machine autonome), mais il semble qu'un seul travailleur python soit créé :

import socket
import threading

spark = SparkSession\
    .builder\
    .master('local[4]')\
    .appName("PythonPi")\
    .getOrCreate()

partitions = 4

# Imprimer l'identifiant du thread local :
print(str(threading.get_ident()))

# Imprimer les identifiants des threads à l'intérieur des travailleurs python :
thread_ids = spark.sparkContext.parallelize(range(1, partitions + 1), partitions)\
.map(lambda x: ' threadid: ' + str(threading.get_ident())).collect()

print(thread_ids)

spark.stop()

Sortie :

140226126948096
[' threadid: 139948131018496', ' threadid: 139948131018496', ' threadid: 139948131018496', ' threadid: 139948131018496']

En examinant ces identifiants de thread, il semblerait que le même thread python (dans le même travailleur) ait été utilisé pour traiter toutes les partitions ? Ou ce code est-il évalué en dehors des travailleurs python ?

Y a-t-il un autre moyen d'accéder à un ID pour les travailleurs python - pour que je puisse comprendre où le code s'exécute ?

Answer 1

Votre erreur est de croire que PySpark utilise le threading. Ce n'est pas le cas. Il utilise des processus et les identifiants de thread en général sont uniques seulement dans un processus (et peuvent être réutilisés).

Donc votre code devrait être :

import os

(spark.sparkContext.range(partitions)
    .map(lambda x: 'pid: {}'.format(os.getpid()))
    .collect())

# ['pid: 749', 'pid: 755', 'pid: 753', 'pid: 758']