28

sparkjava.com フレームワークを使用して、Apache Spark ジョブ用の Web API を構築しようとしています。私のコードは次のとおりです。

@Override
public void init() {
    get("/hello",
            (req, res) -> {
                String sourcePath = "hdfs://spark:54310/input/*";

                SparkConf conf = new SparkConf().setAppName("LineCount");
                conf.setJars(new String[] { "/home/sam/resin-4.0.42/webapps/test.war" });
                File configFile = new File("config.properties");

                String sparkURI = "spark://hamrah:7077";

                conf.setMaster(sparkURI);
                conf.set("spark.driver.allowMultipleContexts", "true");
                JavaSparkContext sc = new JavaSparkContext(conf);

                @SuppressWarnings("resource")
                JavaRDD<String> log = sc.textFile(sourcePath);

                JavaRDD<String> lines = log.filter(x -> {
                    return true;
                });

                return lines.count();
            });
}

ラムダ式を削除するか、Web サービス (何らかの形でサーブレット) ではなく単純な jar 内に配置すると、エラーなしで実行されます。ただし、サーブレット内でラムダ式を使用すると、次の例外が発生します。

15/01/28 10:36:33 WARN TaskSetManager: Lost task 0.0 in stage 0.0 (TID 0, hamrah): java.lang.ClassCastException: cannot assign instance of java.lang.invoke.SerializedLambda to field org.apache.spark.api.java.JavaRDD$$anonfun$filter$1.f$1 of type org.apache.spark.api.java.function.Function in instance of org.apache.spark.api.java.JavaRDD$$anonfun$filter$1
at java.io.ObjectStreamClass$FieldReflector.setObjFieldValues(ObjectStreamClass.java:2089)
at java.io.ObjectStreamClass.setObjFieldValues(ObjectStreamClass.java:1261)
at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1999)
at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1918)
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1801)
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1351)
at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1993)
at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1918)
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1801)
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1351)
at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1993)
at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1918)
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1801)
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1351)
at java.io.ObjectInputStream.readObject(ObjectInputStream.java:371)
at org.apache.spark.serializer.JavaDeserializationStream.readObject(JavaSerializer.scala:62)
at org.apache.spark.serializer.JavaSerializerInstance.deserialize(JavaSerializer.scala:87)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:57)
at org.apache.spark.scheduler.Task.run(Task.scala:56)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:196)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)

PS: jersey と javaspark を jetty、tomcat、および resin と組み合わせてみましたが、すべて同じ結果になりました。

4

4 に答える 4

55

ここにあるのは、元のエラーをマスクするフォローアップ エラーです。

ラムダ インスタンスがシリアライズさwriteReplaceれると、インスタンスである永続的な形式から JRE 固有の実装を解消するために使用されSerializedLambda ます。SerializedLambdaインスタンスが復元されると、そのメソッドreadResolveが呼び出され、適切なラムダ インスタンスが再構成されます。ドキュメントにあるように、元のラムダを定義したクラスの特別なメソッドを呼び出すことでこれを行います (この回答も参照してください)。重要な点は、元のクラスが必要であり、それがあなたの場合に欠けていることです。

しかし、 の…特別な…動作がありObjectInputStreamます。例外が発生しても、すぐには救済されません。例外を記録し、プロセスを続行し、現在読み取られているすべてのオブジェクトをマークするため、エラーのあるオブジェクトもエラーであると見なされます。プロセスの最後にのみ、発生した元の例外をスローします。非常に奇妙なのは、これらのオブジェクトのフィールドを設定しようとし続けることです。ObjectInputStream.readOrdinaryObjectしかし、メソッド行 1806を見ると、次のようになります。

…
    if (obj != null &&
        handles.lookupException(passHandle) == null &&
        desc.hasReadResolveMethod())
    {
        Object rep = desc.invokeReadResolve(obj);
        if (unshared && rep.getClass().isArray()) {
            rep = cloneArray(rep);
        }
        if (rep != obj) {
            handles.setObject(passHandle, obj = rep);
        }
    }

    return obj;
}

非例外を報告するreadResolveときにメソッドを呼び出さないことがわかります。しかし、置換が行われなかった場合、リファラーのフィールド値を設定しようとし続けるのは良い考えではありませんが、それはまさにここで起こっていることなので、.lookupExceptionnullClassCastException

問題を簡単に再現できます。

public class Holder implements Serializable {
    Runnable r;
}
public class Defining {
    public static Holder get() {
        final Holder holder = new Holder();
        holder.r=(Runnable&Serializable)()->{};
        return holder;
    }
}
public class Writing {
    static final File f=new File(System.getProperty("java.io.tmpdir"), "x.ser");
    public static void main(String... arg) throws IOException {
        try(FileOutputStream os=new FileOutputStream(f);
            ObjectOutputStream   oos=new ObjectOutputStream(os)) {
            oos.writeObject(Defining.get());
        }
        System.out.println("written to "+f);
    }
}
public class Reading {
    static final File f=new File(System.getProperty("java.io.tmpdir"), "x.ser");
    public static void main(String... arg) throws IOException, ClassNotFoundException {
        try(FileInputStream is=new FileInputStream(f);
            ObjectInputStream ois=new ObjectInputStream(is)) {
            Holder h=(Holder)ois.readObject();
            System.out.println(h.r);
            h.r.run();
        }
        System.out.println("read from "+f);
    }
}

これら 4 つのクラスをコンパイルして実行しますWriting。次に、クラス ファイルDefining.classを削除して実行しますReading。次に、あなたは

Exception in thread "main" java.lang.ClassCastException: cannot assign instance of java.lang.invoke.SerializedLambda to field test.Holder.r of type java.lang.Runnable in instance of test.Holder
    at java.io.ObjectStreamClass$FieldReflector.setObjFieldValues(ObjectStreamClass.java:2089)
    at java.io.ObjectStreamClass.setObjFieldValues(ObjectStreamClass.java:1261)

(1.8.0_20でテスト済み)

肝心なのは、何が起こっているのかを理解したら、このシリアライゼーションの問題を忘れる可能性があるということです。問題を解決するために必要なのは、ラムダ式を定義したクラスが、ラムダが存在するランタイムでも利用できることを確認することだけです。デシリアライズ。

IDE から直接実行する Spark ジョブの例 (spark-submit はデフォルトで jar を配布します):

SparkConf sconf = new SparkConf()
  .set("spark.eventLog.dir", "hdfs://nn:8020/user/spark/applicationHistory")
  .set("spark.eventLog.enabled", "true")
  .setJars(new String[]{"/path/to/jar/with/your/class.jar"})
  .setMaster("spark://spark.standalone.uri:7077");
于 2015-02-06T14:09:35.440 に答える
4

同じエラーが発生し、ラムダを内部クラスに置き換えたところ、機能しました。理由がよくわかりません。このエラーを再現するのは非常に困難でした (動作を示すサーバーが 1 つだけあり、他にはありませんでした)。

動作しません

this.variable = () -> { ..... }

収量java.lang.ClassCastException: cannot assign instance of java.lang.invoke.SerializedLambda to field MyObject.val$variable

作品

this.variable = new MyInterface() {
    public void myMethod() {
       .....
    }
};
于 2015-09-04T17:21:06.767 に答える
0

Java8ラムダをより簡単に置き換えることができますspark.scala.Function

交換

output = rdds.map(x->this.function(x)).collect()

と：

output = rdds.map(new Function<Double,Double>(){

   public Double call(Double x){
       return MyClass.this.function(x);
   }

}).collect();
于 2017-08-03T14:51:59.253 に答える