Spark 시작하기07 - Java 와 Spark 와 Serializable

Java 혹은 Scala로 람다를 작성시 Serializable Exception 이 자주 발생한다.

컴파일이 아닌, 런타임에러가 발생하므로,,, 도통 원인을 모르겠다.

하나하나 따져보자.

보통 자바 웹개발을 하는 사람은 Serializable을 볼 일이 거의 없다.

네트워크 소켓통신으로 객체를 주고받는 경우가 없기 때문이다.(직렬화)

보통 하는 일이 객체를 JSON 이나 XML로 변경하는 정도??

그래서 더욱... Serializable Exception이 낯설다.

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 rdd2 = sc.parallelize(Arrays.asList(1, 2, 3, 4));
		JavaRDD rdd3 = rdd2.map(a -> a + 1);
		System.out.println(rdd3.collect());
	}

	private void proc2() {
		JavaRDD rdd2 = sc.parallelize(Arrays.asList(1, 2, 3, 4));
		int num2 = 3;
		JavaRDD rdd3 = rdd2.map(a -> a + num2);
		System.out.println(rdd3.collect());
	}
}

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

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 rdd2 = sc.parallelize(Arrays.asList(1, 2, 3, 4));
		JavaRDD rdd3 = rdd2.map(a -> a + this.num1); 				// Exception 발생
		System.out.println(rdd3.collect());
	}

}

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 rdd2 = sc.parallelize(Arrays.asList(1, 2, 3, 4));
		int num1 = this.num1;										// 해결
		JavaRDD rdd3 = rdd2.map(a -> a + num1); 			// 해결
		System.out.println(rdd3.collect());
	}
}

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

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 rdd2 = sc.parallelize(Arrays.asList(1, 2, 3, 4));
		JavaRDD rdd3 = rdd2.map(a -> a + this.num1); 				// 여전히 Exception 발생
		System.out.println(rdd3.collect());
	}
}

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