Erreur dans la mise en cache d'un simple RDD avec pyspark alors sans mise en cache le code fonctionne bien (Comment faire une picklable de classe dans un ordinateur portable)

Question

je le code simple suivant provoque une erreur concernant la mise en cache:

trips_in = sc.textFile("trip_data.csv") 
trips = trips_in.map(lambda l: l.split(",")).map(lambda x: parseTrip(x)).cache() 

trips.count() 

La fonction parseTrip() obtient une liste de chaînes et crée et retourne un voyage de classe:

class Trip: 
    def __init__(self, id, duration): 
    self.id = id 
    self.duration = duration 

Je reçois l'erreur juste après l'action count(). Cependant, si je supprime le cache() à la fin de la deuxième ligne tout fonctionne bien. Selon l'erreur, le problème est que le voyage de classe ne peut pas être décapée:

PicklingError: Can't pickle __main__.Trip: attribute lookup __main__.Trip failed 

Alors, comment puis-je faire picklable (si elle est un mot réel)? Notez que j'utilise un bloc-notes Databricks, donc je ne peux pas faire un .py distinct pour la définition de classe pour le rendre picklable.

Answer 1

L'environnement n'affecte pas la réponse - si vous voulez utiliser des classes personnalisées, il doit être importable sur chaque nœud du cluster.

• Pour un seul module, vous pouvez facilement utiliser SparkContext.addPyFile avec l'URL à un GitHub Gist (ou un autre format pris en charge: "file in HDFS (or other Hadoop-supported filesystems), or an HTTP, HTTPS or FTP URI")

  • Créer un point essentiel.
  • Cliquez sur le lien brut et copiez l'URL.

  • Dans votre appel portable:

    sc.addPyFile(raw_gist_url) 
    

• Pour les dépendances complexes que vous distribuer des fichiers d'œufs.

  • Créer Python packageusing setuptools.

    Structure du répertoire:

    . 
    ├── setup.py 
    └── trip 
        └── __init__.py 
    

    fichier de configuration Exemple:

    #!/usr/bin/env python 
    
    from setuptools import setup 
    
    setup(name='trip', 
         version='0.0.1', 
         description='Trip', 
         author='Jane Doe', 
         author_email='jane@example.com', 
         url='https://example.com', 
         packages=['trip'],) 
    

  • Créer un fichier d'oeuf:

    python setup.py bdist_egg 
    

    Cela va créer le répertoire dist avec trip-0.0.1-pyX.Y.egg fichier

  • Aller au tableau de bord Databricks -> Nouveau -> Libary et télécharger le fichier d'oeuf à partir du répertoire dist:

    

  • bibliothèque à la Joindre cluster que vous souhaitez utiliser.

• Enfin, si vous voulez tout est un type d'enregistrement que vous pouvez utiliser namedtuple sans étapes supplémentaires:

  from collections import namedtuple 
  
  Trip = namedtuple('Trip', ['id', 'duration'])