タスクはシリアル化できません: オブジェクトではなくクラスのみでクロージャの外側で関数を呼び出すと java.io.NotSerializableException が発生します 質問する

RDDはSerialisableインターフェースを拡張しているので、これがタスクの失敗の原因ではありません。これは、RDDSparkでシリアル化して回避できるという意味ではありません。NotSerializableException

Spark は分散コンピューティング エンジンであり、その主な抽象化は分散コレクションとして見ることができる復元力のある分散データセット ( RDD ) です。基本的に、RDD の要素はクラスターのノード間で分割されますが、Spark はこれをユーザーから抽象化し、ユーザーが RDD (コレクション) をローカルのコレクションであるかのように操作できるようにします。

あまり詳しく説明するつもりはありませんが、RDD でさまざまな変換 ( map、flatMapなどfilter) を実行する場合、変換コード (クロージャ) は次のようになります。

1. ドライバーノード上でシリアル化され、
2. クラスター内の適切なノードに送信され、
3. デシリアライズされた、
4. そして最後にノード上で実行される

もちろん、これをローカルで実行することもできます (例のように)。ただし、これらのフェーズはすべて (ネットワーク経由での送信を除いて) 発生します。[これにより、本番環境にデプロイする前にバグをキャッチできます]

2 番目のケースでは、testingマップ関数内からクラスで定義されたメソッドを呼び出しています。Spark はこれを認識し、メソッドは単独ではシリアル化できないため、クラス全体 testingをシリアル化しようとします。そのため、別の JVM で実行された場合でもコードは機能します。次の 2 つの可能性があります。

クラス テストをシリアル化可能にして、クラス全体を Spark でシリアル化できるようにします。

import org.apache.spark.{SparkContext,SparkConf}

object Spark {
  val ctx = new SparkContext(new SparkConf().setAppName("test").setMaster("local[*]"))
}

object NOTworking extends App {
  new Test().doIT
}

class Test extends java.io.Serializable {
  val rddList = Spark.ctx.parallelize(List(1,2,3))

  def doIT() =  {
    val after = rddList.map(someFunc)
    after.collect().foreach(println)
  }

  def someFunc(a: Int) = a + 1
}

import org.apache.spark.{SparkContext,SparkConf}

object Spark {
  val ctx = new SparkContext(new SparkConf().setAppName("test").setMaster("local[*]"))
}

object NOTworking extends App {
  new Test().doIT
}

class Test {
  val rddList = Spark.ctx.parallelize(List(1,2,3))

  def doIT() =  {
    val after = rddList.map(someFunc)
    after.collect().foreach(println)
  }

  val someFunc = (a: Int) => a + 1
}

Serialization stack:
    - object not serializable (class: testing, value: testing@2dfe2f00)
    - field (class: testing$$anonfun$1, name: $outer, type: class testing)
    - object (class testing$$anonfun$1, <function1>)