Применить функцию к столбцу на основе другого значения столбца

Question

Я хочу применить функцию к столбцам во фрейме данных.Применяемая функция зависит от значения одного из столбцов во фрейме данных.Функция сопоставления значений доступна в виде карты.

Источник DF:

TAG       Timestamp              Value
TAG1    2019-06-21 01:16:00.0   621.0947
TAG1    2019-06-21 01:16:00.0   621.0947
TAG1    2019-06-21 01:16:00.0   621.0947
TAG1    2019-06-21 01:16:00.0   619.9578
TAG2    2019-06-21 01:29:00.0   767.5475
TAG2    2019-06-21 01:29:00.0   768.9506
TAG2    2019-06-21 01:29:00.0   770.8863
TAG3    2019-06-21 01:16:00.0   203.7457

Карта:

Map(Tag1 -> avg, Tag2 -> max, Tag3 -> min)

Вывод:

TAG Timestamp            Value
TAG1    2019-06-21 01:16:00.0   620.810475  (avg applied for Tag1 values)
TAG2    2019-06-21 01:29:00.0   770.8863    (max applied)
TAG3    2019-06-21 01:16:00.0   203.7457    (min applied)

Я могу добраться до точки, где яЯ могу получить все значения, агрегированные в столбце, где я застрял, чтобы применить функции динамически.

Ничего в рабочем состоянии.Поэтому я подумал, что было бы целесообразно получить значения в виде списка в столбце, а затем попробовать применить функцию.

val grouped = df.groupBy("TAG").agg(collect_list("value") as "value")

Вывод:

TAG Timestamp            Value
TAG1    2019-06-21 01:16:00.0   620.810475  (avg applied for Tag1 values)
TAG2    2019-06-21 01:29:00.0   770.8863    (max applied)
TAG3    2019-06-21 01:16:00.0   203.7457    (min applied)

Answer 1

Вы можете использовать when...otherwise как оператор case

import spark.implicits._
var df = Seq(
  ("TAG1", "2019-06-21 01:16:00.0",621.0947),
  ("TAG1", "2019-06-21 01:16:00.0",621.0947),
  ("TAG1", "2019-06-21 01:16:00.0",621.0947),
  ("TAG1", "2019-06-21 01:16:00.0",619.9578),
  ("TAG2", "2019-06-21 01:29:00.0",767.5475),
  ("TAG2", "2019-06-21 01:29:00.0",768.9506),
  ("TAG2", "2019-06-21 01:29:00.0",770.8863),
  ("TAG3", "2019-06-21 01:16:00.0",203.7457)).toDF("TAG","Timestamp","Value")

df.groupBy(
  "TAG","Timestamp"
).agg(
  when(
    col("TAG") === "TAG1", avg("Value")
  ).otherwise(
    when(col("TAG") === "TAG2", max("Value")).otherwise(min("Value"))
  ).as("Value")
).show()