org.apache.spark.SparkException : 작업이 스칼라에서 직렬화되지 않음

Question

스칼라를 처음 사용했습니다. 나는 스파크에 코드 아래 실행 오전 : 나는 주석 부분을 제거하고 그것을 잘 표시 데이터를 작동 SparkConf.set(c(0), c(1))을 언급

scala> for(line <- sc.textFile("hdfs://ip:8020/property.conf")) 
     { 
      val c = line.split("=") 
      SparkConf.set(c(0), c(1)) 
      //println(c(0)+" "+c(1)) //Commented 
     } 

합니다.

하지만이 경우 매개 변수를 런타임에 SparkConf으로 설정하고 싶습니다. 그러나 그것은 나에게

org.apache.spark.SparkException 그 오류 발생 : 작업 직렬화하지

나에게 뭔가를 제안 해 주시기 바랍니다.

Answer 1

스파크에 대해 이해하는 것이 매우 중요한 한 가지는 분산 환경이라는 것입니다.

RDD은 Resilient Distributed Datasets의 약식입니다. 스파크 RDD의 항목은 일반적으로 partitions으로 나뉘며, 이는 Spark 클러스터의 다양한 노드에 분산되어 있습니다.

당신이 yourRdd.map(a => a.toString) 같은 전화

이 RDD의 map 구현은 첫째, 클로저이 a => a.toString 기능을 포장 한 후이 RDD의 partitions이있는 모든 노드에 보내 그 폐쇄를 직렬화하는 것을 알고있다. 결과의 실제 계산은 해당 노드에서 수행됩니다.

그래서 ... RDD을 다루는 경우 배포판을 RDD api와 평범한 스칼라 API를 혼동하지 않도록주의하십시오.

당신에게있을 것입니다 코드의 조각을 작성하는 권장되는 방법, 당신의 SparkConf.set(c(0), c(1)) 라인 여기

val yourRdd = sc.textFile("hdfs://ip:8020/property.conf")) 

yourRdd.foreach(line => 
    val c = line.split("=") 
    println(c(0) + " " + c(1)) 
) 

, SparkConf는 class 그리고 당신은 일반적으로 classes를 직렬화 할 수 없습니다. class SparkConf에 멤버 함수 set도 호출 할 수 없습니다. classes의 인스턴스를 만들어야합니다. 또한 SparkConf은 serializable 인터페이스를 구현하지 않는 클래스이기 때문에 SparkConf의 인스턴스조차도 직렬화 할 수 없습니다.

보통 SparkConf을 초기화해야하는 SparkContext이 없으면 RDD가 존재하지 않으므로 RDD를 작성하는 데 스파크 RDD를 사용하지 않아야합니다.

하지만이 경우에는 꼭해야한다고 말한 다음 RDD에서 정상적인 스칼라 목록을 얻은 다음 SparkConf를 만드는 데 사용하십시오.

val mySparkConf = new SparkConf() 

val yourRdd = sc.textFile("hdfs://ip:8020/property.conf")) 

val yourList = yourRdd.foreach(line => 
    val c = line.split("=") 
).collect.toList 

yourList.foreach(c => mySparkConf.set(c(0), c(1)))