[scala] 직렬화 할 수없는 작업 : 객체가 아닌 클래스에서만 클로저 외부에서 함수를 호출 할 때 java.io.NotSerializableException

RDD는 직렬화 가능 인터페이스를 확장 하므로 이로 인해 작업이 실패하지 않습니다. 이제 이것이 RDDSpark로 직렬화 하고 피할 수 있음을 의미하지는 않습니다.NotSerializableException

Spark는 분산 컴퓨팅 엔진이며 주요 추상화는 RDD (Resilient Distributed Dataset )이며 분산 컬렉션으로 볼 수 있습니다. 기본적으로 RDD의 요소는 클러스터의 노드에 분할되어 있지만 Spark는이를 사용자로부터 추상화하여 사용자가 마치 RDD (컬렉션)와 마치 로컬 요소처럼 상호 작용할 수 있도록합니다.

너무 많은 세부 사항으로 얻을 수 있지만, 당신은 RDD (에 다른 변환을 실행할 때하지 않기 map, flatMap, filter등), 당신의 변환 코드 (폐쇄)입니다 :

1. 드라이버 노드에서 직렬화
2. 클러스터의 해당 노드로 배송
3. 역 직렬화
4. 마지막으로 노드에서 실행

물론 로컬에서 (예와 같이) 로컬로 실행할 수 있지만 모든 단계 (네트워크를 통한 배송 제외)는 계속 발생합니다. [이를 통해 프로덕션 환경에 배포하기 전에 버그를 발견 할 수 있습니다]

두 번째 경우 testing에는 map 함수 내부 에서 클래스에 정의 된 메소드를 호출하는 것 입니다. Spark는 메소드를 자체적으로 직렬화 할 수 없으므로 Spark는 전체 testing 클래스 를 직렬화하려고 시도 하므로 다른 JVM에서 실행될 때 코드가 계속 작동합니다. 두 가지 가능성이 있습니다.

클래스 테스트를 직렬화 할 수 있도록 Spark에서 전체 클래스를 직렬화 할 수 있습니다.

import org.apache.spark.{SparkContext,SparkConf}

object Spark {
  val ctx = new SparkContext(new SparkConf().setAppName("test").setMaster("local[*]"))
}

object NOTworking extends App {
  new Test().doIT
}

class Test extends java.io.Serializable {
  val rddList = Spark.ctx.parallelize(List(1,2,3))

  def doIT() =  {
    val after = rddList.map(someFunc)
    after.collect().foreach(println)
  }

  def someFunc(a: Int) = a + 1
}

또는 someFuncSpark가 메소드를 직렬화 할 수 있도록 메소드 대신 함수 를 만듭니다 (함수는 스칼라의 객체입니다).

import org.apache.spark.{SparkContext,SparkConf}

object Spark {
  val ctx = new SparkContext(new SparkConf().setAppName("test").setMaster("local[*]"))
}

object NOTworking extends App {
  new Test().doIT
}

class Test {
  val rddList = Spark.ctx.parallelize(List(1,2,3))

  def doIT() =  {
    val after = rddList.map(someFunc)
    after.collect().foreach(println)
  }

  val someFunc = (a: Int) => a + 1
}

비슷하지만 클래스 직렬화와 동일한 문제는 관심이 될 수 있으며이 Spark Summit 2013 프레젠테이션에서 읽을 수 있습니다 .

보조 노트로서, 당신은 다시 작성할 수 있습니다 rddList.map(someFunc(_))에 rddList.map(someFunc)그들이 정확히 동일합니다. 일반적으로 두 번째는 읽기가 덜 장황하고 깨끗하므로 선호됩니다.

편집 (2015-03-15) : SPARK-5307 은 SerializationDebugger를 도입 했으며 Spark 1.3.0은 그것을 사용하는 첫 번째 버전입니다. NotSerializableException에 직렬화 경로를 추가합니다 . NotSerializableException이 발생하면 디버거는 개체 그래프를 방문하여 직렬화 할 수없는 개체의 경로를 찾고 사용자가 개체를 쉽게 찾을 수 있도록 정보를 구성합니다.

OP의 경우 stdout으로 인쇄됩니다.

Serialization stack:
    - object not serializable (class: testing, value: testing@2dfe2f00)
    - field (class: testing$$anonfun$1, name: $outer, type: class testing)
    - object (class testing$$anonfun$1, <function1>)

답변

def genMapper[A, B](f: A => B): A => B = {
  val locker = com.twitter.chill.MeatLocker(f)
  x => locker.get.apply(x)
}

rdd map genMapper(someFunc)

@transient
val rddList = Spark.ctx.parallelize(list)

def genMapper(kryoWrapper: KryoSerializationWrapper[(Foo => Bar)])
               (foo: Foo) : Bar = {
    kryoWrapper.value.apply(foo)
}
val mapper = genMapper(KryoSerializationWrapper(new Blah(abc))) _
rdd.flatMap(mapper).collectAsMap()

object Blah(abc: ABC) extends (Foo => Bar) {
    def apply(foo: Foo) : Bar = { //This is the real function }
}

object NOTworking extends App {
 new testing().doIT
}
//adding extends Serializable wont help
class testing {

val list = List(1,2,3)

val rddList = Spark.ctx.parallelize(list)

def doIT =  {
  //again calling the fucntion someFunc 
  val after = rddList.map(someFunc(_))
  //this will crash (spark lazy)
  after.collect().map(println(_))
}

def someFunc(a:Int) = a+1

}

def doIT =  {
 def someFunc(a:Int) = a+1
  //function definition
 }
 val after = rddList.map(someFunc(_))
 after.collect().map(println(_))
}

spark.kryo.unsafe="false"

spark.serializer="org.apache.spark.serializer.JavaSerializer"

val sampleMap = Map("index1" -> 1234, "index2" -> 2345)
val outputRDD = rdd.map(row => {
    val value = sampleMap.get(row._1)
    value
})

import com.twitter.chill.MeatLocker
val sampleMap = Map("index1" -> 1234, "index2" -> 2345)
val brdSerSampleMap = spark.sparkContext.broadcast(MeatLocker(sampleMap))

rdd.mapPartitions(partition => {
    val deSerSampleMap = brdSerSampleMap.value.get
    partition.map(row => {
        val value = sampleMap.get(row._1)
        value
    }).toIterator
})