PySpark de diffusion variables locales des fonctions

Je suis d'essayer de créer de diffusion des variables à partir de l'intérieur de Python méthodes (essayer de résumé de certaines méthodes de l'utilitaire, je suis de la création qui s'appuient sur des opérations réparties). Cependant, je n'arrive pas à accéder à l'émission des variables à partir de l'intérieur de l'Étincelle travailleurs.

Disons que j'ai cette config:

def main():
    sc = SparkContext()
    SomeMethod(sc)

def SomeMethod(sc):
    someValue = rand()
    V = sc.broadcast(someValue)
    A = sc.parallelize().map(worker)

def worker(element):
    element *= V.value  ### NameError: global name 'V' is not defined ###

Cependant, si j'ai au lieu de supprimer le SomeMethod() intermédiaire, il fonctionne très bien.

def main():
    sc = SparkContext()
    someValue = rand()
    V = sc.broadcast(someValue)
    A = sc.parallelize().map(worker)

def worker(element):
    element *= V.value   # works just fine

Je préfère ne pas mettre tous mes Étincelle logique dans la méthode main, si je peux. Est-il possible de diffuser des variables sur les fonctions et de leur monde visible à l'Étincelle travailleurs?

Sinon, ce serait un bon modèle de conception pour ce genre de situation-par exemple, je veux écrire une méthode spécifiquement pour l'Étincelle qui s'est auto-contenue et exerce une fonction spécifique, je voudrais ré-utiliser?

InformationsquelleAutor Magsol | 2014-11-16
  1. 15

    Je ne suis pas sûr que j'ai bien saisi la question, mais, si vous avez besoin de la V objets à l'intérieur du travailleur en fonction de vous, alors vous avez certainement devraient se passer comme paramètre, sinon la méthode n'est pas vraiment autonome:

    def worker(V, element):
    element *= V.value

    Maintenant afin de l'utiliser dans les fonctions de mappage, vous devez utiliser une partielle, de sorte que la carte ne voit qu'un 1 paramètre fonction:

    from functools import partial

def SomeMethod(sc):
    someValue = rand()
    V = sc.broadcast(someValue)
    A = sc.parallelize().map(partial(worker, V=V))
    • Existe-il des implications sur les performances en passant de diffusion variables comme ça? Dire, par exemple, j'ai été en s'appuyant sur une diffusion variable dans une map() de la fonction sur des dizaines de milliers (ou plus) de lignes. quelque chose comme def transform(row): return broadcast_variable.value[row[0]] qui est ensuite utilisée dans un map() fonction comme rdd.map(transform)
    • Grâce à cette solution m'a aidé à éviter l'utilisation de mondial pour la diffusion de la variable. Veuillez noter que vous devez remplacer l'ordre du travailleur paramètres de la méthode, de sorte que l '"élément" paramètre (qui remplie par Étincelle cadre) seront les premiers. Sinon il ne fonctionnera pas.
    • je vous remercie beaucoup. Vous avez vraiment sauvé ma journée. Je vais avoir un autre problème avec l'étincelle mais votre suggestion dans mon scénario travaillé comme un charme. Un grand MERCI !!
    InformationsquelleAutor elyase