как добиться выполнения нескольких функций в каждом столбце динамически? - PullRequest
Новая
       7

как добиться выполнения нескольких функций в каждом столбце динамически?

1 голос
/ 16 января 2020

Я использую spark- sql -2.4.1v с java8.

У меня есть следующий сценарий 

val df = Seq(
  ("0.9192019",  "0.1992019",  "0.9955999"),
  ("0.9292018",  "0.2992019",  "0.99662018"),
  ("0.9392017",  "0.3992019",  "0.99772000")).toDF("item1_value","item2_value","item3_value")
.withColumn("item1_value", $"item1_value".cast(DoubleType))
.withColumn("item2_value", $"item2_value".cast(DoubleType))
.withColumn("item3_value", $"item3_value".cast(DoubleType))

df.show(20)

Мне нужен ожидаемый вывод что-то вроде этого 

-----------------------------------------------------------------------------------
col_name      |  sum_of_column     | avg_of_column   | vari_of_column 
-----------------------------------------------------------------------------------
"item1_value" | sum("item1_value") | avg("item1_value") | variance("item1_value")
"item2_value" | sum("item2_value") | avg("item2_value") | variance("item2_value")
"item3_value" | sum("item3_value") | avg("item3_value") | variance("item3_value")
----------------------------------------------------------------------------------

как этого добиться динамично .. завтра у меня может быть

1 Ответ

2 голосов
/ 16 января 2020

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

import org.apache.spark.sql.types._
import org.apache.spark.sql.Column

val df = Seq(
  ("0.9192019",  "0.1992019",  "0.9955999"),
  ("0.9292018",  "0.2992019",  "0.99662018"),
  ("0.9392017",  "0.3992019",  "0.99772000")).
  toDF("item1_value","item2_value","item3_value").
  withColumn("item1_value", $"item1_value".cast(DoubleType)).
  withColumn("item2_value", $"item2_value".cast(DoubleType)).
  withColumn("item3_value", $"item3_value".cast(DoubleType))

val aggregateColumns = Seq("item1_value","item2_value","item3_value")

var aggDFs = aggregateColumns.map( c => {
    df.groupBy().agg(lit(c).as("col_name"),sum(c).as("sum_of_column"), avg(c).as("avg_of_column"), variance(c).as("var_of_column"))
})

var combinedDF = aggDFs.reduce(_ union _)

Возвращает следующий вывод: 

scala> df.show(10,false)
+-----------+-----------+-----------+
|item1_value|item2_value|item3_value|
+-----------+-----------+-----------+
|0.9192019  |0.1992019  |0.9955999  |
|0.9292018  |0.2992019  |0.99662018 |
|0.9392017  |0.3992019  |0.99772    |
+-----------+-----------+-----------+


scala> combinedDF.show(10,false)
+-----------+------------------+------------------+---------------------+
|col_name   |sum_of_column     |avg_of_column     |var_of_column        |
+-----------+------------------+------------------+---------------------+
|item1_value|2.7876054         |0.9292018         |9.999800000999957E-5 |
|item2_value|0.8976057000000001|0.2992019         |0.010000000000000002 |
|item3_value|2.9899400800000002|0.9966466933333334|1.1242332201333484E-6|
+-----------+------------------+------------------+---------------------+
...