apache spark에서 클래스 (작업 객체)의 kryo 직렬화가 역 직렬화 중에 null을 반환합니다.

Question

Java Spark API를 사용하여 일부 테스트 응용 프로그램을 작성하고 있습니다. serializable 인터페이스를 확장하지 않는 클래스를 사용하고 있습니다. 그래서 응용 프로그램을 작동 시키려면 kryo serializer를 사용하여 클래스를 직렬화합니다. 하지만 내가 디버깅하는 동안 관찰 한 문제는 디 직렬화 중에 반환 된 클래스 객체는 null이되어 결국 null 포인터 예외을 던졌습니다. 상황이 잘못되었지만 확실하지 않은 폐쇄 문제 인 것 같습니다. 이러한 직렬화에 익숙하지 않아서 어디에서 파기를 시작할 지 모르겠습니다. 내가 직렬화하고있는 클래스

다음

package org.apache.spark.examples; 


import java.io.FileWriter; 
import java.io.IOException; 
import java.io.PrintWriter; 
import java.net.InetAddress; 
import java.net.UnknownHostException; 

import org.apache.spark.SparkConf; 
import org.apache.spark.api.java.JavaRDD; 
import org.apache.spark.api.java.JavaSparkContext; 
import org.apache.spark.api.java.function.Function; 




/** 
* Spark application to test the Serialization issue in spark 
*/ 
public class Test { 

    static PrintWriter outputFileWriter; 
    static FileWriter file; 
    static JavaSparkContext ssc; 

    public static void main(String[] args) { 


     String inputFile = "/home/incubator-spark/examples/src/main/scala/org/apache/spark/examples/InputFile.txt"; 

     String master = "local"; 
     String jobName = "TestSerialization"; 
     String sparkHome = "/home/test/Spark_Installation/spark-0.7.0"; 
     String sparkJar = "/home/test/TestSerializationIssesInSpark/TestSparkSerIssueApp/target/TestSparkSerIssueApp-0.0.1-SNAPSHOT.jar"; 


     SparkConf conf = new SparkConf(); 
     conf.set("spark.closure.serializer","org.apache.spark.serializer.KryoSerializer"); 
     conf.set("spark.kryo.registrator", "org.apache.spark.examples.MyRegistrator"); 
     // create the Spark context 
     if(master.equals("local")){ 
      ssc = new JavaSparkContext("local", jobName,conf); 
      //ssc = new JavaSparkContext("local", jobName); 
     } else { 
      ssc = new JavaSparkContext(master, jobName, sparkHome, sparkJar); 
     } 
     JavaRDD<String> testData = ssc.textFile(inputFile).cache(); 
     final NotSerializableJavaClass notSerializableTestObject= new NotSerializableJavaClass("Hi "); 
     @SuppressWarnings({ "serial", "unchecked"}) 
     JavaRDD<String> classificationResults = testData.map(
       new Function<String, String>() { 
        @Override 
        public String call(String inputRecord) throws Exception {     
         if(!inputRecord.isEmpty()) { 
          //String[] pointDimensions = inputRecord.split(","); 
          String result = ""; 

          try { 
           FileWriter file = new FileWriter("/home/test/TestSerializationIssesInSpark/results/test_result_" + (int) (Math.random() * 100)); 
           PrintWriter outputFile = new PrintWriter(file); 
           InetAddress ip; 
           ip = InetAddress.getLocalHost(); 
           outputFile.println("IP of the server: " + ip); 

           result = notSerializableTestObject.testMethod(inputRecord); 
           outputFile.println("Result: " + result); 

           outputFile.flush(); 
           outputFile.close(); 
           file.close(); 

          } catch (UnknownHostException e) { 
           e.printStackTrace(); 
          } 
          catch (IOException e1) { 
           e1.printStackTrace(); 
          } 

          return result; 
         } else { 
          System.out.println("End of elements in the stream."); 
          String result = "End of elements in the input data"; 
          return result; 
         } 
        } 

       }).cache(); 

     long processedRecords = classificationResults.count(); 

     ssc.stop(); 
     System.out.println("sssssssssss"+processedRecords); 
    } 
} 

인 KryoRegistrator 클래스 다음

package org.apache.spark.examples; 

import org.apache.spark.serializer.KryoRegistrator; 

import com.esotericsoftware.kryo.Kryo; 

public class MyRegistrator implements KryoRegistrator { 
    public void registerClasses(Kryo kryo) { 
     kryo.register(NotSerializableJavaClass.class); 
    } 
} 

됩니다 :

package org.apache.spark.examples; 

public class NotSerializableJavaClass { 
    public String testVariable; 

    public NotSerializableJavaClass(String testVariable) { 
     super(); 
     this.testVariable = testVariable; 
    } 

    public String testMethod(String vartoAppend){ 
     return this.testVariable + vartoAppend; 
    } 
} 

다음

내가 테스트입니다 코드입니다

Answer 1

spark.closure.serializer은 Java serializer 만 지원하기 때문입니다. http://spark.apache.org/docs/latest/configuration.html 약 spark.closure.serializer