Làm thế nào để xử lý RDDs bằng cách sử dụng một lớp Python?

Question

Tôi đang triển khai một mô hình trong Spark dưới dạng lớp python và bất kỳ lúc nào tôi cố ánh xạ phương thức lớp tới RDD không thành công. mã thực tế của tôi là phức tạp hơn, nhưng phiên bản đơn giản này được ở trung tâm của vấn đề:

class model(object): 
    def __init__(self): 
     self.data = sc.textFile('path/to/data.csv') 
     # other misc setup 
    def run_model(self): 
     self.data = self.data.map(self.transformation_function) 
    def transformation_function(self,row): 
     row = row.split(',') 
     return row[0]+row[1] 

Bây giờ, nếu tôi chạy mô hình như vậy (ví dụ):

test = model() 
test.run_model() 
test.data.take(10) 

tôi nhận được lỗi sau:

Ngoại lệ: Có vẻ như bạn đang cố gắng tham khảo SparkContext từ biến phát sóng, hành động hoặc chuyển đổi. SparkContext chỉ có thể được sử dụng trên trình điều khiển, không phải trong mã mà nó chạy trên công nhân. Để biết thêm thông tin, xem SPARK-5063.

Tôi đã chơi với điều này một chút, và nó dường như đáng tin cậy xảy ra bất cứ lúc nào tôi cố gắng để ánh xạ một phương pháp lớp vào một RDD trong lớp. Tôi đã xác nhận rằng các chức năng ánh xạ hoạt động tốt nếu tôi thực hiện bên ngoài của một cấu trúc lớp học, do đó, vấn đề chắc chắn đã làm với lớp. Có cách nào để giải quyết vấn đề này không?

Answer 1

Vấn đề ở đây hơi tinh tế hơn một chút so với sử dụng các RDD lồng nhau hoặc thực hiện các hành động Spark bên trong các phép biến đổi. Spark không cho phép truy cập vào hành động hoặc chuyển đổi bên trong SparkContext.

Thậm chí bạn không truy cập rõ ràng nó được tham chiếu bên trong phần đóng và phải được tuần tự hóa và mang theo. Nó có nghĩa là phương pháp transformation của bạn, tham chiếu self, giữ SparkContext là tốt, do đó lỗi.

Một cách để xử lý này là sử dụng phương pháp tĩnh:

class model(object): 
    @staticmethod 
    def transformation_function(row): 
     row = row.split(',') 
     return row[0]+row[1] 

    def __init__(self): 
     self.data = sc.textFile('some.csv') 

    def run_model(self): 
     self.data = self.data.map(model.transformation_function) 

Sửa:

Nếu bạn muốn để có thể truy cập các biến Ví dụ bạn có thể thử một cái gì đó như thế này:

class model(object): 
    @staticmethod 
    def transformation_function(a_model): 
     delim = a_model.delim 
     def _transformation_function(row): 
      return row.split(delim) 
     return _transformation_function 

    def __init__(self): 
     self.delim = ',' 
     self.data = sc.textFile('some.csv') 

    def run_model(self): 
     self.data = self.data.map(model.transformation_function(self))