Apache Spark RDDタスクのシリアル化の理解

Question

私は、タスクのシリアル化がSparkでどのように機能するのかを理解しようとしていますが、私が書いたテストでは混乱している結果が少しあります。

私は複数のノード上で、以下のん（ポストのために簡略化され）、いくつかのテストコードがあります：私はrun()を実行すると、「タスク直列化可能ではない」を除いて、アウトジョブ爆弾、

object TestJob { 
    def run(): Unit = { 
    val rdd = ... 
    val helperObject = new Helper() // Helper does NOT impl Serializable and is a vanilla class 
    rdd.map(element => { 
     helperObject.transform(element) 
    }).collect() 
    } 
} 

をhelperObjectはシリアライズ可能ではないため、期待通りです。しかし、私が少し変更すると、次のようになります。

trait HelperComponent { 
    val helperObject = new Helper() 
} 

object TestJob extends HelperComponent { 
    def run(): Unit = { 
    val rdd = ... 
    rdd.map(element => { 
     helperObject.transform(element) 
    }).collect() 
    } 
} 

何らかの理由でジョブが正常に実行されます。なぜ誰かがこのことが理解できるように助けてくれますか？ Sparkによって正確にシリアル化され、上記の各ケースでワーカーに送信されるものは何ですか？

私はSparkバージョン2.1.1を使用しています。

ありがとうございました！

Answer 1

これがなぜ起こったのか理解してくれる人がいますか？

最初のスニペットでは、helperObjectはrunの中で宣言されたローカル変数です。このように、このコードが実行される場所であれば、すべての情報が利用可能になり、そのためにSparks ClosureCleanerがシリアル化しようとしてあなたに叫ぶように、関数によって閉じられます（持ち上げられます）。

2番目のスニペットでは、値はメソッドスコープのローカル変数ではなく、クラスインスタンスの一部です（技術的にはこれはオブジェクト宣言ですが、結局はJVMクラスに変換されます）。 ）。

クラスタ内のすべてのワーカーノードにコードを実行するために必要なJARが含まれているため、これはSparkで意味があります。したがって、のためにTestObjectの全体をシリアライズする代わりに、SparkがあなたのワーカーのExecutorプロセスをスピンアップするとき、ClassLoaderを介してローカルにTestObjectをロードし、それ以外のJVMクラスと同じようにインスタンスを作成します分散アプリケーション。

結論として、あなたがタイプインスタンスを宣言した方法の変更のために、クラスがもはやシリアル化されていないため、この爆発が見られない理由があります。