スパーク作業者間のコード共有

Question

スパークの文書によれば、RDD変換では匿名またはスカラのオブジェクト関数を使用する方がよいでしょう。私は次のコードを持つオブジェクトがあります。

object Util { 
    val someManager = new Manager() 

    def process(data: String) = someManager.manage(data) 
} 

を私は次の方法でそれを呼び出す：

Utilのオブジェクトは、労働者を刺激するためにシリアライズして送信されますどのように

myRDD.map(Util.process) 

？マネージャーは毎回または一度だけ送信するように作成されますか？どのくらいの時間、マネージャのインスタンスがスパーク作業員に送られますか？

Answer 1

Utilオブジェクトをシリアル化してスパーク作業者に送信する方法はありますか。 毎回または1回だけ送信するマネージャが作成されますか？どのくらい 時間マネージャのインスタンスがスパーク作業員に送信されますか？

Util.processは、それをシリアル化する内容であるapply方法、だ実装し、コンパイル時にAbstractFunction1内にラップされます。あなたは、このコードをコンパイルするときは、（これはJVMバイトコードから再構築された）、これを見ることができます：

par.map(new AbstractFunction1() { 
    public static final long serialVersionUID = 0L; 

    public final void apply(String data) 
    { 
    Util..MODULE$.process(data); 
    } 
}, ClassTag..MODULE$.Unit()); 
} 

を我々が見たいもう一つがどのようにUtilオブジェクトの取得のバイトコードにコンパイル：

public final class Util$ 
{ 
    public static final MODULE$; 
    private final Manager someManager; 

    public Manager someManager() 
    { 
    return this.someManager; 
    } 

    private Util$() 
    { 
    MODULE$ = this; 
    this.someManager = new Manager(); 
    } 

    public void process(String data) 
    { 
    someManager().manage(data); 
    } 
} 

つまり、まだManagerのインスタンスフィールドがあります。何が起きるかは、それが必要ではないので、ClosureCleanerをUtilのインスタンスを削除するために使用しますが、インスタンスをワーカーにシリアル化する必要があります。毎回mapを呼び出す必要があります。これは、Managerの複数のインスタンスが存在することを意味するものではなく、単一のインスタンスをシリアル化してワイヤを介して送信する必要があることを意味します。メカニズムが実際に始まるの閉鎖によって を使用されていない道に沿ってすべての参照を囲むクロージャの階層とヌルを横断するある

• 、しかし：私たちは、ClosureCleanerのドキュメントで詳細を確認することができますそれにもかかわらず、コンパイルされた匿名クラス に含まれています。他のコードパスが に依存する可能性があるので、 が囲みクロージャを所定位置に変更するのは安全ではないことに注意してください。代わりに、それぞれの囲みクロージャをクローンし、それに応じて 親ポインタを設定します。

Answer 2

Utilはシリアライズ可能である必要があります。そうでないと例外が発生します。マネージャは、毎回、例えばすべてのタスクで送信されます。パーティションごとに一度だけ送信することができます。これは、最初にブロードキャストする必要があるためです。