如何使用Python类处理RDD？

Question

我正在Spark中将模型实现为python类，并且每次尝试将类方法映射到RDD时，它都会失败。我的实际代码更加复杂，但是这个简化的版本成为了问题的核心：

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]

现在，如果我像这样运行模型（例如）：

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

我收到以下错误：

例外：看来您试图从广播变量，操作或转换中引用SparkContext。SparkContext只能在驱动程序上使用，而不能在工作程序上运行的代码中使用。有关更多信息，请参见SPARK-5063。

我已经玩了一点，并且似乎在我尝试将类方法映射到类中的RDD时，确实可以可靠地发生。我已经确认，如果我在类结构之外实现映射函数，则该函数可以正常工作，因此问题肯定与该类有关。有办法解决吗？

Answer 1

与使用 嵌套的RDD或在转换内部执行Spark操作 相比，这里的问题要微妙得多。Spark不允许访问SparkContext内部操作或转换。

即使您没有显式访问它，也要在闭包内部对其进行引用，并且必须对其进行序列化和携带。这意味着transformation引用的方法也会self保留SparkContext，因此会出现错误。

一种解决方法是使用静态方法：

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)

编辑 ：

如果您希望能够访问实例变量，可以尝试如下操作：

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))