Добиться динамической c интерполяции строк в Scala искре? - PullRequest
Новая
       8

Добиться динамической c интерполяции строк в Scala искре?

0 голосов
/ 19 января 2020

У меня есть строка с функциями, которые должны go в .agg функции моего предполагаемого кадра данных. Мой фрейм данных выглядит как 

val client = Seq((1,"A","D",10),(2,"A","D",5),(3,"B","C",56),(5,"B","D",67)).toDF("ID","Categ","subCat","Amnt")
+---+-----+------+----+
| ID|Categ|subCat|Amnt|
+---+-----+------+----+
|  1|    A|     D|  10|
|  2|    A|     D|   5|
|  3|    B|     C|  56|
|  5|    B|     D|  67|
+---+-----+------+----+

, поэтому я пытаюсь интерполировать этот тип 

val str= "s"$count(ID) as Total,$sum(Amnt) as amt""

Я хочу добиться этого как выходной 

client.groupBy("Categ","subCat").agg(sum("Amnt") as "amt",count("ID") as "Total").show()
+-----+------+---+-----+
|Categ|subCat|amt|Total|
+-----+------+---+-----+
|    B|     C| 56|    1|
|    A|     D| 15|    2|
|    B|     D| 67|    1|
+-----+------+---+-----+

Я попробовал это 

 client.groupBy("Categ","subCat").agg(s"$str").show()

получаю ошибку 

> error: overloaded method value agg with alternatives:

(expr: org. apache .spark. sql .Column, exprs: org. apache .spark. sql .Column *) org. apache .spark. sql .DataFrame
(exprs: java .util.Map [String, String]) org. apache .spark. sql .DataFrame (exprs: scala .collection.immutable.Map [String, String]) org. apache .spark. sql .DataFrame (aggExpr: (String, String), aggExprs: (String, String) *) org. apache .spark. sql .DataFrame нельзя применить к (String)

Я также пробовал expr 

    val str="sum(Amnt) as amt"
    client.groupBy("Categ","subCat").agg(expr(str)).show()

 this return the desired outcome

    +-----+------+---+
    |Categ|subCat|amt|
    +-----+------+---+
    |    B|     C| 56|
    |    A|     D| 15|
    |    B|     D| 67|
    +-----+------+---+

но снова, когда я пытаюсь val str="sum(Amnt) as amt,count(ID) as ID_tot"

    client.groupBy("Categ","subCat").agg(expr(str)).show()
org.apache.spark.sql.catalyst.parser.ParseException:
mismatched input ',' expecting <EOF>(line 1, pos 16)

Ответы [ 2 ]

3 голосов
/ 19 января 2020

Немного грубое решение: разделить на , и вызвать expr для каждого: 

val str="sum(Amnt) as amt,count(ID) as ID_tot"
val (first, rest) = str.split(",").map(expr).splitAt(1)
client.groupBy("Categ","subCat").agg(first, rest: _*)

Если , может быть частью выражения (например, внутри строкового литерала), оно становится хуже: попробуйте разобрать его с expr, поймать ParseException и посмотреть, чем он заканчивается? Должен быть более простой путь, но я этого не знаю.

3 голосов
/ 19 января 2020

Вы можете использовать другой API для достижения желаемого 

import org.apache.spark.sql.functions._
client
  .groupBy("Categ", "subCat")
  .agg(
    sum("Amnt").as("Amnt"), 
    count(lit("1")).as("Total"))
  .show()

В качестве альтернативы, go full Spark SQL, но я бы не рекомендовал этот шаблон, если вы пытаетесь ввести ввод пользователя. 

spark.sql(s"select Categ, subCat, $str from client group by Categ, subCat")
...