스파크 작업이 직렬화되지 않음

StackOverflow에서 찾은이 문제에 대한 모든 솔루션을 시도했지만이 문제는 해결할 수 없습니다. "Recommendation"개체를 인스턴스화하는 "MainObj"개체가 있습니다. "recommendationProducts"메서드를 호출하면 항상 오류가 발생합니다. 다음은 메소드의 코드는 다음과 같습니다 ") 1)"직렬화 확장 "(스칼라)를 내 클래스 2 : 내가 추가하려고스파크 작업이 직렬화되지 않음

Exception in thread "main" org.apache.spark.SparkException: Task not serializable 
at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:298) 
at org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:288) 
at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:108) 
at org.apache.spark.SparkContext.clean(SparkContext.scala:2094) 
at org.apache.spark.rdd.RDD$$anonfun$map$1.apply(RDD.scala:370) 
at org.apache.spark.rdd.RDD$$anonfun$map$1.apply(RDD.scala:369) 
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151) 
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112) 
at org.apache.spark.rdd.RDD.withScope(RDD.scala:362) 
at org.apache.spark.rdd.RDD.map(RDD.scala:369) 
at RecommendationObj.recommendationProducts(RecommendationObj.scala:269) 
at MainObj$.analisiIUNGO(MainObj.scala:257) 
at MainObj$.menu(MainObj.scala:54) 
at MainObj$.main(MainObj.scala:37) 
at MainObj.main(MainObj.scala) 
Caused by: java.io.NotSerializableException: org.apache.spark.SparkContext 
Serialization stack: 
- object not serializable (class: org.apache.spark.SparkContext, value: [email protected]) 
- field (class: RecommendationObj, name: sc, type: class org.apache.spark.SparkContext) 
- object (class MainObj$$anon$1, [email protected]) 
- field (class: RecommendationObj$$anonfun$37, name: $outer, type: class RecommendationObj) 
- object (class RecommendationObj$$anonfun$37, <function1>) 
at org.apache.spark.serializer.SerializationDebugger$.improveException(SerializationDebugger.scala:40) 
at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:46) 
at org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:100) 
at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:295) 
... 14 more

def recommendationProducts(item: Int): Unit = { 

val aMatrix = new DoubleMatrix(Array(1.0, 2.0, 3.0)) 

def cosineSimilarity(vec1: DoubleMatrix, vec2: DoubleMatrix): Double = { 
    vec1.dot(vec2)/(vec1.norm2() * vec2.norm2()) 
} 

val itemFactor = model.productFeatures.lookup(item).head 
val itemVector = new DoubleMatrix(itemFactor) 

//Here is where I get the error: 
val sims = model.productFeatures.map { case (id, factor) => 
    val factorVector = new DoubleMatrix(factor) 
    val sim = cosineSimilarity(factorVector, itemVector) 
    (id, sim) 
} 

val sortedSims = sims.top(10)(Ordering.by[(Int, Double), Double] { 
    case (id, similarity) => similarity 
}) 

println("\nTop 10 products:") 
sortedSims.map(x => (x._1, x._2)).foreach(println)

이

오류입니다 확장 java.io.Serializable "을 내 클래스 3)"@transient "to some parts 4)이 클래스 내부에서 모델 (및 다른 기능)을 가져옵니다. (이제 다른 객체에서 가져 와서 클래스와 같은 인수)

어떻게 해결할 수 있습니까? 나는 미쳐 가고있다! 미리 감사드립니다.

출처

2017-10-25 S. SP

키는 여기에 있습니다 :

field (class: RecommendationObj, name: sc, type: class org.apache.spark.SparkContext)

그래서 당신이 형 SparkContext의 필드라는 이름의 SC있다. Spark은 클래스를 직렬화하려고하므로 모든 필드를 직렬화하려고 시도합니다.

당신이해야 :

사용 @Transient 주석 및 널 (null), 다음
필드에서 SparkContext를 사용하지 다시, 그러나 방법의 인수에 넣어 경우 검사합니다. 그러나 맵, 플랫 맵 등의 클로저에서 SparkContext를 사용해서는 안됩니다.

출처

2017-10-25 13:41:05

감사합니다! 그것은 작동합니다! 그러나 클래스의 인수와 같은 SparkContext (sc)를 전달하고이를 생성자에서 모델을로드하는 데 사용합니다. 잘못 됐나? –

@SSP 당신을 도운다면 투표를하고 대답을 수락하십시오. 잘못된 것은 아니지만 serializer가 serialize하지 않도록 '@transient'주석을 사용해야합니다. –

Ok! 고맙습니다. 나는 당신에게 투표를하고 싶지만 아직도 그것을하기위한 15 가지 평판 포인트가 없습니다. 너의 대답이 나에게 너무 많은 도움이 되었기 때문에 미안해! –

스파크 작업이 직렬화되지 않음

답변

관련 문제