Pythonクラスを使ってRDDを処理する方法は？

Question

私は、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) 

私が取得します次のエラー：

例外：それはあなたが放送変数、アクション、またはtransforamtionからSparkContextを参照しようとしていることが表示されます。 SparkContextはドライバでのみ使用でき、ワーカーで実行されるコードでは使用できません。詳細は、SPARK-5063を参照してください。

私はこれで少しプレイしました、そして、確実に私はクラス内RDDにクラスメソッドをマップしようといつでも発生するようです。クラス構造の外部に実装すると、マップされた関数がうまく動作することが確認されています。したがって、この問題は間違いなくクラスと関係しています。これを解決する方法はありますか？

Answer 1

ここでは、ネストされたRDDを使用するか、変換内でSparkアクションを実行するよりも少し微妙です。 Sparkでは、アクションまたはトランスフォーメーション内のSparkContextへのアクセスを許可していません。

でもあなたはそれが明示的にそれがクロージャ内で参照され、シリアライズして持ち歩くことをしているアクセスしないでください。それはselfを参照して、あなたのtransformation方法は、誤差したがって、同様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))