[SCALA] 불꽃 DataFrame에있는 객체의 배열을 받아들이는 UDF를 정의?

불꽃 DataFrame에있는 객체의 배열을 받아들이는 UDF를 정의?

스파크의 DataFrames 작업 할 때, 사용자 정의 함수 (UDF)의 열에 매핑 데이터가 필요합니다. UDF는 그 인수의 형식을 명시 적으로 지정해야합니다. 내 경우, 나는 개체의 배열로 구성되어 열을 조작해야, 내가 사용 유형을 알 수 없습니다. 다음은 그 예이다 :

import sqlContext.implicits._

// Start with some data. Each row (here, there's only one row) 
// is a topic and a bunch of subjects
val data = sqlContext.read.json(sc.parallelize(Seq(
  """
  |{
  |  "topic" : "pets",
  |  "subjects" : [
  |    {"type" : "cat", "score" : 10},
  |    {"type" : "dog", "score" : 1}
  |  ]
  |}
  """)))

이 컬럼의 데이터에 대한 기본 작업을 수행 할 수 내장 된 org.apache.spark.sql.functions를 사용하는 비교적 간단합니다

import org.apache.spark.sql.functions.size
data.select($"topic", size($"subjects")).show

+-----+--------------+
|topic|size(subjects)|
+-----+--------------+
| pets|             2|
+-----+--------------+

그것은 임의의 작업을 수행하는 사용자 정의 UDF를 작성하는 일반적으로 쉽게

import org.apache.spark.sql.functions.udf
val enhance = udf { topic : String => topic.toUpperCase() }
data.select(enhance($"topic"), size($"subjects")).show 

+----------+--------------+
|UDF(topic)|size(subjects)|
+----------+--------------+
|      PETS|             2|
+----------+--------------+

그러나 나는 "대상"열에서 객체의 배열을 조작 할 수 UDF를 사용하려면? 나는 UDF의 인수에 대해 어떤 종류를 사용합니까? 예를 들어, I 대신 스파크에 의해 제공되는 하나를 사용하는 크기의 기능을 구현할 것인지 :

val my_size = udf { subjects: Array[Something] => subjects.size }
data.select($"topic", my_size($"subjects")).show

명확하게 배열 [뭔가] 유형을 사용해야합니까 ... 작동하지 않습니다!? 나는 모두 배열 [] 도랑해야 하는가? 주위를 파고 것은 scala.collection.mutable.WrappedArray이 함께 할 수있는 뭔가가있을 수 있지만, 아직도 내가 제공해야 또 다른 유형이있다 나에게 말한다.

해결법

1. ==============================

   1.당신이 찾고있는 서열 [o.a.s.sql.Row]입니다 :

   당신이 찾고있는 서열 [o.a.s.sql.Row]입니다 :

   import org.apache.spark.sql.Row
   
   val my_size = udf { subjects: Seq[Row] => subjects.size }
   

   설명:

   노트:

from https://stackoverflow.com/questions/39006349/defining-a-udf-that-accepts-an-array-of-objects-in-a-spark-dataframe by cc-by-sa and MIT license