[SCALA] UDF 함수는 충분히 큰 입력 변수에 동의하지 스파크 DataFrames

UDF 함수는 충분히 큰 입력 변수에 동의하지 스파크 DataFrames

나는 ID와 DataFrame을 준비하고 내 기능의 벡터 예측을하고 나중에 사용할 수 있습니다. 내 dataframe에 GROUPBY을, 내 GROUPBY에 나는 새로운 열에 목록 등 열 몇 가지를 병합하고 있습니다 :

def mergeFunction(...) // with 14 input variables

val myudffunction( mergeFunction ) // Spark doesn't support this

df.groupBy("id").agg(
   collect_list(df(...)) as ...
   ... // too many of these (something like 14 of them)
).withColumn("features_labels",
  myudffunction(
     col(...)
     , col(...) )
.select("id", "feature_labels")

내 특징 벡터 및 자신의 레이블을 만드는 오전 방법입니다. 그것은 지금까지 나를 위해 일하고있다 그러나 이것은이 방법 나의 특징 벡터가 최대 스파크에서 UDF 함수가 받아들이는 것입니다 수 (10)보다 큰지고 것은 이번이 처음이다.

해결법

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

   1.사용자 정의 함수는 22 개의 매개 변수를 정의합니다. 만 도우미 UDF 최대 10 개 인수에 대한 정의입니다. 당신이 org.apache.spark.sql.UDFRegistration를 사용할 수있는 매개 변수의 많은 수와 기능을 처리합니다.

   사용자 정의 함수는 22 개의 매개 변수를 정의합니다. 만 도우미 UDF 최대 10 개 인수에 대한 정의입니다. 당신이 org.apache.spark.sql.UDFRegistration를 사용할 수있는 매개 변수의 많은 수와 기능을 처리합니다.

   예를 들면

   val dummy = ((
     x0: Int, x1: Int, x2: Int, x3: Int, x4: Int, x5: Int, x6: Int, x7: Int, 
     x8: Int, x9: Int, x10: Int, x11: Int, x12: Int, x13: Int, x14: Int, 
     x15: Int, x16: Int, x17: Int, x18: Int, x19: Int, x20: Int, x21: Int) => 1)
   

   에 의해 등록 할 수 :

   import org.apache.spark.sql.expressions.UserDefinedFunction
   
   val dummyUdf: UserDefinedFunction = spark.udf.register("dummy", dummy)
   

   직접 사용

   val df = spark.range(1)
   val exprs =  (0 to 21).map(_ => lit(1))
   
   df.select(dummyUdf(exprs: _*))
   

   또는 callUdf를 통해 이름으로

   import org.apache.spark.sql.functions.callUDF
   
   df.select(
     callUDF("dummy", exprs:  _*).alias("dummy")
   )
   

   또는 SQL 표현 :

   df.selectExpr(s"""dummy(${Seq.fill(22)(1).mkString(",")})""")
   

   당신은 또한 UserDefinedFunction 개체를 만들 수 있습니다 :

   import org.apache.spark.sql.expressions.UserDefinedFunction
   
   Seq(1).toDF.select(UserDefinedFunction(dummy, IntegerType, None)(exprs: _*))
   

   실제로 22 개 인수 기능을 갖는 것은 매우 유용하지 않습니다 당신이 생성하는 스칼라 반사를 사용하지 않으려면이 유지 보수 악몽이있다.

   I는 어느 입력으로서 모음 (배열 MAP) 또는 구조체의 사용을 고려하거나 여러 모듈로 분할이있다. 예를 들면 :

   val aLongArray = array((0 to 256).map(_ => lit(1)): _*)
   
   val udfWitharray = udf((xs: Seq[Int]) => 1)
   
   Seq(1).toDF.select(udfWitharray(aLongArray).alias("dummy"))
   

2. ==============================

   2.그냥 제로의 대답을 확장하기 위해서는 10 개 이상의 매개 변수가있는 UDF와 함께 작동하도록 .withColumn () 함수를 얻을 수 있습니다. spark.udf.register에 그냥 필요 () 다음 기능과 (대신 UDF의) 열을 추가 인수에 대한 EXPR를 사용합니다.

   그냥 제로의 대답을 확장하기 위해서는 10 개 이상의 매개 변수가있는 UDF와 함께 작동하도록 .withColumn () 함수를 얻을 수 있습니다. spark.udf.register에 그냥 필요 () 다음 기능과 (대신 UDF의) 열을 추가 인수에 대한 EXPR를 사용합니다.

   예를 들어, 다음과 같이 작동합니다 :

   def mergeFunction(...) // with 14 input variables
   
   spark.udf.register("mergeFunction", mergeFunction) // make available in expressions
   
   df.groupBy("id").agg(
      collect_list(df(...)) as ...
      ... // too many of these (something like 14 of them)
   ).withColumn("features_labels",
     expr("mergeFunction(col1, col2, col3, col4, ...)") ) //pass in the 14 column names
   .select("id", "feature_labels")
   

   난 당신이 함수를 호출하는 배열을 주변에 전달에 의존 할 필요가 있다고 생각하지 않도록 근본적인 표현 파서는 10 개 이상의 매개 변수를 처리하는 것 같다. 그들은 매개 변수를 다른 데이터 유형으로 발생하는 경우 또한, 배열은 매우 잘 작동하지 않을 것입니다.

from https://stackoverflow.com/questions/39494620/spark-dataframes-when-udf-functions-do-not-accept-large-enough-input-variables by cc-by-sa and MIT license