Iterieren benutzerdefinierte Klassenobjekte in einem RD RD RD

Question

Ich lese Daten aus CSV und konvertiert diese Daten in ein Python-Objekt. Aber wenn ich versuche, dass mit benutzerdefinierten Objekte Klasse RDD iterieren, erhalte ich Fehler wie,

_pickle.PicklingError: Can't pickle <class '__main__.User'>: attribute lookup User on __main__ failed 

ich hier einen Teil des Codes hinzufügen,

class User: 
    def __init__(self, line): 
     self.user_id = line[0] 
     self.location = line[1] 
     self.age = line[2] 

def create_user(line): 
    user = User(line) 
    return user 

def print_user(line): 
    user = line 
    print(user.user_id) 

conf = (SparkConf().setMaster("local").setAppName("exercise_set_2").set("spark.executor.memory", "1g")) 
sc = SparkContext(conf = conf) 
users = sc.textFile("BX-Users.csv").map(lambda line: line.split(";")) 
users_objs = users.map(lambda entry: create_user(entry)) 
users_objs.map(lambda entry: print_user(entry)) 

Für den obigen Code , erhalte ich Ergebnisse wie,

PythonRDD[93] at RDD at PythonRDD.scala:43 

CSV-Datenquelle URL (eine Zip-Extraktion benötigt): HERE

UPDATE: Wenn ich den Code ändere, um Collect einzufügen, wird es wieder zu einem Fehler kommen. Ich muss es trotzdem mit Pickle versuchen. Ich habe das vorher noch nie ausprobiert. Wenn jemand eine Probe hat, kann ich es leicht machen.

users_objs = users.map(lambda entry: create_user(entry)).collect() 

Answer 1

Okay, eine Erklärung gefunden. Durch das Speichern von Klassen in separaten Dateien werden die Klassen automatisch einfügbar gemacht. Also habe ich die User-Klasse in user.py gespeichert und den folgenden Import in meinen Code eingefügt.

from user import User 

Inhalt User.py

class User: 
    def __init__(self, line): 
     self.user_id = line[0] 
     self.location = line[1] 
     self.age = line[2] 

Wie in früheren Antwort erwähnte, kann ich Benutzer (eine RDD-Methode) auf den erstellten Benutzerobjekte sammeln. Der folgende Code wird also alle Benutzer-IDs drucken, wie ich es wollte.

for user_obj in users.map(lambda entry: create_user(entry)).collect(): 
    print_user(user_obj) 

Answer 2

Wenn Sie

verwenden

def create_user(line): 
    user = User(line) 
    return user 

direkt in einer Karte Anruf, bedeutet dies, dass die User Klasse Ihren Knoten zugänglich sein muss. Typischerweise bedeutet dies, dass es serialisierbar/pickbar sein muss. Wie würde ein Knoten diese Klasse verwenden oder wissen, was sie ist (es sei denn, Sie haben eine gemeinsame NFS-Bereitstellung oder etwas Ähnliches)? Deshalb haben Sie diesen Essiggurkenfehler bekommen. Um Ihre User Klasse einfügbar zu machen, lesen Sie bitte diese: https://docs.python.org/2/library/pickle.html.

Darüber hinaus führen Sie keine collect() auf Ihrer RDD, weshalb Sie PythonRDD[93] at RDD at PythonRDD.scala:43 sehen. Es ist immer noch eine RDD, Ihre Daten sind auf den Knoten.