Spark - оптимизировать запрос с несколькими «когда» - PullRequest
Новая
       29

Spark - оптимизировать запрос с несколькими «когда»

0 голосов
/ 07 ноября 2018

Есть ли способ оптимизировать этот запрос, чтобы не использовать withColumn несколько раз. Моя самая большая проблема в том, что я столкнулся с этой проблемой: https://issues.apache.org/jira/browse/SPARK-18532

Запрос примерно такой. У меня есть датафрейм с 10 булевыми столбцами. У меня есть несколько модификаторов, таких как: 

val smallIncrease = 5
val smallDecrease = -5
val bigIncrease = 10
val bigDecrease = -10

Исходя из каждого логического столбца, я хотел бы рассчитать итоговую оценку, добавив небольшое / большое увеличение / уменьшение на основе значений в разных столбцах. Теперь мой запрос выглядит примерно так: 

df.withColumn("result", when(col("col1"), col("result") + lit(smallIncrease)).otherwise(col("result") + lit(smallDecrease)))
.withColumn("result", when(col("col2"), col("result") + lit(bigIncrease)).otherwise(col("result") + lit(bigDecrease)))
.withColumn("result", when(col("col3"), col("result") + lit(smallIncrease)).otherwise(col("result") + lit(smallDecrease)))
.withColumn("result", when(col("col4"), col("result") + lit(smallIncrease)).otherwise(col("result") + lit(smallDecrease)))
.withColumn("result", when(col("col5"), col("result") + lit(smallIncrease)).otherwise(col("result") + lit(bigDecrease)))
.withColumn("result", when(col("col6"), col("result") + lit(bigIncrease)).otherwise(col("result") + lit(smallDecrease)))
.withColumn("result", when(col("col7"), col("result") + lit(smallIncrease)).otherwise(col("result") + lit(smallDecrease)))

Есть ли способ сжать этот запрос и избежать нескольких withColumns. UDF к сожалению не подходит, так как нужно учитывать более 10 логических столбцов, а UDF ограничены 10 столбцами. Может быть, я могу разделить его на 2 UDF, но для меня это выглядит ужасно ...

1 Ответ

0 голосов
/ 07 ноября 2018

Как насчет этого? 

def myFun(b: Seq[Boolean], result: Int): Int = {

  val conversions: Seq[(Boolean, Int) => Int] = ??? // Functions to apply increase/decrease for each boolean value col1, col2 etc.

  b.zip(conversions).foldLeft(result){
    case (acc, (nextBool, nextFun)) => nextFun(nextBool, acc) 
  }
}

val myUdf = udf(myFun(_: Seq[Boolean], _: Int))

df.select(myUdf(array($"col1", $"col2", $"col3"...), $"result").as("result"))
...