Cloudera CDH/CDP 및 Hadoop EcoSystem, Semantic IoT등의 개발/운영 기술을 정리합니다. gooper@gooper.com로 문의 주세요.

spark Spark에서 Serializable관련 오류및 조치사항

총관리자 2017.04.21 17:40 조회 수 : 5287

1. 성공케이스

package org.mystudy.testcase;
 
import java.util.Arrays;
 
import org.apache.log4j.PropertyConfigurator;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
 
public class TestCase1 {
    JavaSparkContext sc = null;
 
    private TestCase1() {
        PropertyConfigurator.configure("D:\workspace\spark\learning.spark\src\resources\log4j.properties");
        sc = new JavaSparkContext("local[2]", "First Spark App");
    }
 
    public static void main(String... strings) {
        TestCase1 t = new TestCase1();
        t.proc1();
        t.proc2();
    }
 
    private void proc1() {
        JavaRDD<integer> rdd2 = sc.parallelize(Arrays.asList(1, 2, 3, 4));
        JavaRDD<integer> rdd3 = rdd2.map(a -> a + 1);
        System.out.println(rdd3.collect());
    }
 
    private void proc2() {
        JavaRDD<integer> rdd2 = sc.parallelize(Arrays.asList(1, 2, 3, 4));
        int num2 = 3;
        JavaRDD<integer> rdd3 = rdd2.map(a -> a + num2);
        System.out.println(rdd3.collect());
    }
}
 
</integer></integer></integer></integer>

좋은 케이스 : 에러 없이 잘... 작동한다.

JAVA8의 람다식이다.

2. 실패사례 - 전역변수(멤버필드)

package org.mystudy.testcase;
 
import java.util.Arrays;
 
import org.apache.log4j.PropertyConfigurator;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
 
public class TestCase2 {
    private int num1 = 4;
    JavaSparkContext sc = null;
 
    private TestCase2() {
        PropertyConfigurator.configure("D:\workspace\spark\learning.spark\src\resources\log4j.properties");
        sc = new JavaSparkContext("local[2]", "First Spark App");
    }
 
    public static void main(String... strings) {
        TestCase2 t = new TestCase2();
        System.out.println("t:"+t);
        t.proc3();
    }
 
    private void proc3() {
        JavaRDD<integer> rdd2 = sc.parallelize(Arrays.asList(1, 2, 3, 4));
        JavaRDD<integer> rdd3 = rdd2.map(a -> a + this.num1);                // Exception 발생
        System.out.println(rdd3.collect());
    }
 
}
 
 
</integer></integer>

Exception 발생

람다식에 this.num1 이 사용되었다. this는 TestCase2 자체를 의미하므로, 현재 TestCase2 가 Serializable 을 구현하지 않았으므로 아래와 같은 Exception 이 발생한다.

16/04/08 00:01:10 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable

t:org.mystudy.testcase.TestCase2@247667dd

Exception in thread "main" org.apache.spark.SparkException: Task not serializable

at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:304)

at org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:294)

at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:122)

at org.apache.spark.SparkContext.clean(SparkContext.scala:2055)

at org.apache.spark.rdd.RDD$$anonfun$map$1.apply(RDD.scala:324)

at org.apache.spark.rdd.RDD$$anonfun$map$1.apply(RDD.scala:323)

at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:150)

at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:111)

at org.apache.spark.rdd.RDD.withScope(RDD.scala:316)

at org.apache.spark.rdd.RDD.map(RDD.scala:323)

at org.apache.spark.api.java.JavaRDDLike$class.map(JavaRDDLike.scala:96)

at org.apache.spark.api.java.AbstractJavaRDDLike.map(JavaRDDLike.scala:46)

at org.mystudy.testcase.TestCase2.proc3(TestCase2.java:26)

at org.mystudy.testcase.TestCase2.main(TestCase2.java:21)

Caused by: java.io.NotSerializableException: org.mystudy.testcase.TestCase2

Serialization stack:

- object not serializable (class: org.mystudy.testcase.TestCase2, value: org.mystudy.testcase.TestCase2@247667dd)

- element of array (index: 0)

- array (class [Ljava.lang.Object;, size 1)

- field (class: java.lang.invoke.SerializedLambda, name: capturedArgs, type: class [Ljava.lang.Object;)

- object (class java.lang.invoke.SerializedLambda, SerializedLambda[capturingClass=class org.mystudy.testcase.TestCase2, functionalInterfaceMethod=org/apache/spark/api/java/function/Function.call:(Ljava/lang/Object;)Ljava/lang/Object;, implementation=invokeSpecial org/mystudy/testcase/TestCase2.lambda$0:(Ljava/lang/Integer;)Ljava/lang/Integer;, instantiatedMethodType=(Ljava/lang/Integer;)Ljava/lang/Integer;, numCaptured=1])

- writeReplace data (class: java.lang.invoke.SerializedLambda)

- object (class org.mystudy.testcase.TestCase2$$Lambda$4/503353142, org.mystudy.testcase.TestCase2$$Lambda$4/503353142@7a1f8def)

- field (class: org.apache.spark.api.java.JavaPairRDD$$anonfun$toScalaFunction$1, name: fun$1, type: interface org.apache.spark.api.java.function.Function)

- object (class org.apache.spark.api.java.JavaPairRDD$$anonfun$toScalaFunction$1, <function1>)

at org.apache.spark.serializer.SerializationDebugger$.improveException(SerializationDebugger.scala:40)

at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:47)

at org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:101)

at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:301)

... 13 more

2-1 해결책

package org.mystudy.testcase;
 
import java.util.Arrays;
 
import org.apache.log4j.PropertyConfigurator;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
 
public class TestCase2Sol1 {
    private int num1 = 4;
    JavaSparkContext sc = null;
 
    private TestCase2Sol1() {
        PropertyConfigurator.configure("D:\workspace\spark\learning.spark\src\resources\log4j.properties");
        sc = new JavaSparkContext("local[2]", "First Spark App");
    }
 
    public static void main(String... strings) {
        TestCase2Sol1 t = new TestCase2Sol1();
        t.proc3();
    }
 
    private void proc3() {
        JavaRDD<integer> rdd2 = sc.parallelize(Arrays.asList(1, 2, 3, 4));
        int num1 = this.num1;                                       // 해결
        JavaRDD<integer> rdd3 = rdd2.map(a -> a + num1);             // 해결
        System.out.println(rdd3.collect());
    }
}
 
</integer></integer>

[러닝 스파크] 책에서 소개하는 방식으로...

this.num1의 값을 지역변수로 재할당해서 사용하면 된다.

2-2 이렇게도 해결할 수 있을까? 안돼~

package org.mystudy.testcase;
 
import java.io.Serializable;
import java.util.Arrays;
 
import org.apache.log4j.PropertyConfigurator;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
 
public class TestCase2Sol2 implements Serializable {
    private int num1 = 4;
    private JavaSparkContext sc = null;
 
    private TestCase2Sol2() {
        PropertyConfigurator.configure("D:\workspace\spark\learning.spark\src\resources\log4j.properties");
        sc = new JavaSparkContext("local[2]", "First Spark App");
    }
 
    public static void main(String... strings) {
        TestCase2Sol2 t = new TestCase2Sol2();
        System.out.println("t:"+t);
        System.out.println("sc:"+t.sc);
        t.proc3();
    }
 
    private void proc3() {
        JavaRDD<integer> rdd2 = sc.parallelize(Arrays.asList(1, 2, 3, 4));
        JavaRDD<integer> rdd3 = rdd2.map(a -> a + this.num1);                // 여전히 Exception 발생
        System.out.println(rdd3.collect());
    }
}
 
 
</integer></integer>

implements Serializable 을 했음에도 Exception이 발생한다.

이유인즉은, JavaSparkContext 객체를 위 코드에서 클래스의 전역변수로 사용하고 있는데, 아무리 클래스에 Serializable을 구현해놓아도

멤버필드 즉, JavaSparkContext sc 는 기본적으로 직렬화가 안되는 모양이다;;;

16/04/08 00:10:33 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable

t:org.mystudy.testcase.TestCase2Sol2@247667dd

sc:org.apache.spark.api.java.JavaSparkContext@6f099cef