org.apache.spark.SparkException：スカラーでタスクがシリアライズできない

Question

私はスカラを初めて利用しています。私は、Spark上のコードの下で実行しています：

scala> for(line <- sc.textFile("hdfs://ip:8020/property.conf")) 
     { 
      val c = line.split("=") 
      SparkConf.set(c(0), c(1)) 
      //println(c(0)+" "+c(1)) //Commented 
     } 

私はコメント部分とコメントSparkConf.set(c(0), c(1))を削除した場合、それは罰金や表示データを動作します。

しかし、この場合、実行時にパラメータをSparkConfに設定したいとします。 しかし、それは

org.apache.spark.SparkExceptionことを私にエラーがスローされます。私に何かを提案してください

タスク直列化可能ではありません。

Answer 1

スパークについて理解する上で非常に重要なことの1つは、分散環境だということです。

RDDの名前はResilient Distributed Datasetsの略語です。スパークRDD内の項目は、一般的に、partitionsに分割され、Sparkクラスタのさまざまなノードに分散されます。

あなたがyourRdd.map(a => a.toString)のようなものを呼び出すと、このRDDのmap実装は、それは最初、閉鎖にこのa => a.toString機能をラップし、その閉鎖をシリアル化し、このRDDのpartitionsを持っているすべてのノードに送信しなければならないことを知っています。結果の実際の計算はそれらのノードで行われます。

RDDさんと対処しているときには、配布に対応するRDD apiを通常のScala APIと混同しないように注意してください。

コードの一部がなるあなたを書くために推奨される方法、

val yourRdd = sc.textFile("hdfs://ip:8020/property.conf")) 

yourRdd.foreach(line => 
    val c = line.split("=") 
    println(c(0) + " " + c(1)) 
) 

ここにあなたのSparkConf.set(c(0), c(1))ラインで、SparkConfはclassであり、あなたは、通常classesをシリアル化することはできません。 class SparkConfにメンバー関数setも呼び出すことはできません。 classesのインスタンスを作成する必要があります。また、SparkConfは、シリアル化可能なインタフェースを実装していないクラスであるため、SparkConfのインスタンスさえもシリアル化できません。

SparkConfを作成するのに通常は火花RDDを使用しないでください。SparkContextがなくてもRDDが存在しないため、SparkConfを初期化する必要があります。

しかし、この場合、あなたはそれを行う必要があると言うことができます。次に、RDDから通常のスカラーリストを取得してからSparkConfを作成します。

val mySparkConf = new SparkConf() 

val yourRdd = sc.textFile("hdfs://ip:8020/property.conf")) 

val yourList = yourRdd.foreach(line => 
    val c = line.split("=") 
).collect.toList 

yourList.foreach(c => mySparkConf.set(c(0), c(1)))