Apache Zeppelin에서 udf를 적용 할 때 클래스를 찾을 수 없음

Question

나중에 편집 : Apache Zeppelin 인터프리터와 관련된 문제인 것 같습니다. Spark 1.6.0에서 Apache Zeppelin 0.6.0을 사용하고 있습니다. spark-shell (2.0.0)에서 같은 코드를 실행할 때 아무런 문제가 없었습니다.

이것은 너무 구체적 일지 모르지만 UDF로 유사한 오류가 발생하는 다른 사람들을 도울 수 있습니다.

내가 원하는 것은 해당 DF의 다른 열과 문자열의 Seq를 기반으로 Spark Dataframe에 열을 만드는 것입니다. 따라서 "도시"열을 만들고 "위치"열의 값이 0 인 다른 도시 "도시"에 있으면 1을 입력하십시오.

여러 가지 방법으로 시도해 보았습니다. 같은 오류가 발생합니다. 최종 버전은 다음 게시물을 기반으로합니다. Use of Seq.contains(String) 및 Create new column with udf. 이것은 내가 지금 무엇을 가지고 : 내가 받기를 실행하면

val cities = Seq("london", "paris") 
df.filter(lower($"location") isin (cities : _*)).count() 

긴 5485947 그래서 내가 그 2 곳

import org.apache.spark.sql.functions._ 
val urbanFlag: (String => Int) = (arg: String) => {if (cities.contains(arg)) 1 else 0} 
val urbf = udf(urbanFlag) 
df.withColumn("urban", urbf(lower($"location"))).show(100) 

와 기록이 = "작업으로 인해 무대 장애로 중단"오류 :

java.lang.ClassNotFoundException: $iwC$$iwC$$iwC$$iwC$$iwC$$$$725d9ae18728ec9520b65ad133e3b55$$$$$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$anonfun$1

... 거대한 스택 추적. 익명 함수에 대해 뭔가있는 것 같아요?

Answer 1

아마도 UDF를 정의하는 방식에 문제가 있습니까? 이 작품은 나를 위해 :

import org.apache.spark.sql.functions._ 

val data = sqlContext.read.json(sc.parallelize(Seq("{'location' : 'london'}", "{'location': 'tokyo'}"))) 

val cities = Seq("london", "paris") 
val urbf = udf { city: String => if (cities.contains(city)) 1 else 0 } 

data.select($"location", urbf($"location")).show 

+--------+-------------+ 
|location|UDF(location)| 
+--------+-------------+ 
| london|   1| 
| tokyo|   0| 
+--------+-------------+ 

나는 중급없이 직접 UDF를 정의하고있다.