Когда условие в groupBy функции искры sql

Question

Я хочу применить, если условие в операции groupBy искрового кадра данных.Если первое условие выполнено, выберите столбец «A», в противном случае столбец «B» данного кадра данных

Проще вернуть один столбец в условие столбца groupBy.

Например,

df.groupBy(when(col("name") === "a",col("city")).otherwise(col("country"))).agg(lit("Individual").alias("level")).show

Приведенный выше код дает мне результат.Но если я хочу вернуть несколько столбцов в зависимости от условия if, то он не работает

Мой код:

val df = Seq(
  ("a", "abcdef", "123" ,"def", "uyhiu"),
  ("a", "7yjbb", "345" ,"hgh", "hjjhj"),
  ("d", "sbkbnn", "456","gyu", "hghj" )
).toDF("name", "email", "phone", "city", "country")

   val list1 = Array("phone", "city")
   val list2 = Array("phone", "country")

df.groupBy(when(col("name") === "a",list1.map(col): _*).otherwise(list2.map(col):_*)).agg(lit("Individual").alias("level")).show

Но я получаю ошибку:

:52: ошибка: нет: _ * 'здесь разрешены аннотации (такие аннотации допускаются только в аргументах -параметров) df.groupBy (when (col ("name") === "a", list1.map (col): _ ). в противном случае (list2.map (столбец): _ )). agg (горит («Индивидуальный»). псевдоним («уровень»)). show ^: 52: ошибка: нет: _ 'здесь разрешены аннотации (такие аннотации допускаются только в аргументах для -параметров) df.groupBy (когда (col ("name") === "a", list1.map (col):_ ). В противном случае (list2.map (столбец): _ *)). Agg (горит («Индивидуальный»). Псевдоним («уровень»)). Show

Answer 1

Мне кажется, что вы используете неверный подход.Вы не можете динамически изменять имена столбцов предложения groupBy для каждой записи.Это может быть результатом некоторого выражения, но не манипулирования именами столбцов.Вы можете использовать фильтры и сделать объединение позже, как показано ниже.

scala> val df = Seq(
     |   ("a", "abcdef", "123" ,"def", "uyhiu"),
     |   ("a", "7yjbb", "345" ,"hgh", "hjjhj"),
     |   ("d", "sbkbnn", "456","gyu", "hghj" )
     | ).toDF("name", "email", "phone", "city", "country")
df: org.apache.spark.sql.DataFrame = [name: string, email: string ... 3 more fields]

scala>  val list1 = Array("phone", "city")
list1: Array[String] = Array(phone, city)

scala> val list2 = Array("phone", "country")
list2: Array[String] = Array(phone, country)

scala> val df1 = df.filter("name='a'").groupBy(list1.map(col(_)):_*).agg(lit("Individual").alias("level"))
df1: org.apache.spark.sql.DataFrame = [phone: string, city: string ... 1 more field]

scala> val df2 = df.filter("name!='a'").groupBy(list2.map(col(_)):_*).agg(lit("Individual").alias("level"))
df2: org.apache.spark.sql.DataFrame = [phone: string, country: string ... 1 more field]

scala> df1.union(df2).show
+-----+----+----------+
|phone|city|     level|
+-----+----+----------+
|  345| hgh|Individual|
|  123| def|Individual|
|  456|hghj|Individual|
+-----+----+----------+


scala>

Answer 2

Вы должны применить выражение when к обоим столбцам:

df.groupBy(
  when(col("name") === "a", col("phone")).otherwise(col("city")),
  when(col("name") === "a", col("phone")).otherwise(col("country"))
)

Конечно, вы можете предварительно построить их, используя некоторые операции сбора:

val names = Vector(("phone", "city"), ("phone", "country"))

val columns = names.map {
  case (ifTrue, ifFalse) =>
    when(col("name") === "a", col(ifTrue)).otherwise(col(ifFalse))
}

df.groupBy(columns: _*)