스파크 dataframe는 : 유형 장치에 대한 스키마

Question

내가 스파크 1.5.0을 사용하고 지원되지 않으며 나는이 문제를 가지고 : 당신이 열 설명 이있다 볼 수

다음

val df = paired_rdd.reduceByKey { 
    case (val1, val2) => val1 + "|" + val2 
}.toDF("user_id","description") 

이, DF에 대한 샘플 데이터입니다 형식 (text1#text3#weight | text1#text3#weight|....)

사용자 1

book1#author1#0.07841217886795074|tool1#desc1#0.27044260397331488|song1#album1#-0.052661673730870676|item1#category1#-0.005683148395350108

여기 내림차순으로 중량을 기준으로 DF이를 정렬 할

내가 뭘하려 :

먼저 "|" 다음 해당 문자열의 각각에 대해, "#"그들을 분할 무게 3 문자열을 얻을 다음 내림차순으로 (이중 값으로 UDF에 의해 반환 된 무게 값을 기준으로

val getSplitAtWeight = udf((str: String) => { 
    str.split("|").foreach(_.split("#")(2).toDouble) 
}) 

정렬하는 변환 방식)

val df_sorted = df.sort(getSplitAtWeight(col("description")).desc) 

나는 다음과 같은 오류 얻을 : 당신의 udf에

Exception in thread "main" java.lang.UnsupportedOperationException: Schema for type Unit is not supported at org.apache.spark.sql.catalyst.ScalaReflection$class.schemaFor(ScalaReflection.scala:153) at org.apache.spark.sql.catalyst.ScalaReflection$.schemaFor(ScalaReflection.scala:29) at org.apache.spark.sql.catalyst.ScalaReflection$class.schemaFor(ScalaReflection.scala:64) at org.apache.spark.sql.catalyst.ScalaReflection$.schemaFor(ScalaReflection.scala:29) at org.apache.spark.sql.functions$.udf(functions.scala:2242)

Answer 1

변경 foreach을예외를 제거합니다 다음과 같이 당신의 방법

def getSplitAtWeight = udf((str: String) => { 
    str.split('|').map(_.split('#')(2).toDouble) 
}) 

문제는 List에 foreach 방법이 무엇이든, 즉, 그 결과는 당신이 Exception을받을 이유 유형 Unit이다를 반환하지 않는다는 것입니다. foreach에 대한 자세한 내용을 보려면 this blog을 확인하십시오.