Spark Scala Serializableオブジェクトを動的に作成する

Question

Scala Sparkフィルタにはテスターを使用し、テスターはjavaのPredicateインターフェースを実装し、引数で特定のクラス名を受け取る必要があります。 私はこの

val tester = Class.forName(qualifiedName).newInstance().asInstanceOf[Predicate[T]] 
var filtered = rdd.filter(elem => tester.test(elem)) 

のようなものをやっている問題は、私の特定の述語クラスがシリアライズではないので、実行時に、私はスパーク「TaskNotSerializable例外」を持っているということです。

私は

val tester = Class.forName(qualifiedName).newInstance() 
      .asInstanceOf[Predicate[T] with Serializable] 
var filtered = rdd.filter(elem => tester.test(elem)) 

をすれば、私は同じエラーを取得します。 私はrdd.filterにテスターを作成する場合はそれが動作を呼び出す：

var filtered = rdd.filter { elem => 
    val tester = Class.forName(qualifiedName).newInstance() 
      .asInstanceOf[Predicate[T] with Serializable] 
    tester.test(elem) 
} 

しかし、私はテストのために（多分放送するために）単一のオブジェクトを作成します。どうすれば解決できますか？

Answer 1

クラスの実装には、Serializableが必要です。 asInstanceOf[Predicate[T] with Serializable]キャストは嘘です：実際に値がSerializableであることを確認していないので、キャスト中に2番目のケースでエラーが発生せず、最後のケースでは「成功」する理由があります。

しかし、テストのために1つのオブジェクト（ブロードキャストする可能性があります）を作成します。

できません。ブロードキャストであるかどうかにかかわらず、デシリアライゼーションは、という新しいワーカーノード上のオブジェクトを作成します。しかし、あなたは、各パーティションにのみ単一のインスタンスを作成することができます。

var filtered = rdd.mapPartitions { iter => 
    val tester = Class.forName(qualifiedName).newInstance() 
      .asInstanceOf[Predicate[T]] 
    iter.filter(tester.test) 
} 

をそれは実際にそれが厳密に小さい仕事だから、それを送信し、それを考えデシリアライズ、testerをシリアル化するよりもパフォーマンスが向上します。