спарк динамически создать структуру / JSON на группу

Question

У меня есть искровой фрейм данных, такой как

+-----+---+---+---+------+
|group|  a|  b|  c|config|
+-----+---+---+---+------+
|    a|  1|  2|  3|   [a]|
|    b|  2|  3|  4|[a, b]|
+-----+---+---+---+------+
val df = Seq(("a", 1, 2, 3, Seq("a")),("b", 2, 3,4, Seq("a", "b"))).toDF("group", "a", "b","c", "config")

Как добавить дополнительный столбец, например

df.withColumn("select_by_config", <<>>).show

в виде структуры или JSON, который объединяет несколько столбцов (определяется config) в чем-то похожем на улей с именем struct / spark struct / json column?Обратите внимание, что эта структура специфична для каждой группы и не является постоянной для всего фрейма данных;это указано в столбце config.

Я могу себе представить, что df.map может сработать, но издержки сериализации не кажутся эффективными.Как это может быть достигнуто с помощью выражений только SQL?Может быть, в качестве столбца типа карты?

edit

возможное, но действительно неуклюжее решение для 2.2:

val df = Seq((1,"a", 1, 2, 3, Seq("a")),(2, "b", 2, 3,4, Seq("a", "b"))).toDF("id", "group", "a", "b","c", "config")
  df.show
  import spark.implicits._
  final case class Foo(id:Int, c1:Int, specific:Map[String, Int])
  df.map(r => {
    val config = r.getAs[Seq[String]]("config")
    print(config)
    val others = config.map(elem => (elem, r.getAs[Int](elem))).toMap
    Foo(r.getAs[Int]("id"), r.getAs[Int]("c"), others)
  }).show

существуют ли более эффективные способы решения проблемы для2,2

Answer 1

Если вы используете недавнюю сборку (Spark 2.4.0 RC 1 или новее), комбинация функций более высокого порядка должна помочь.Создайте карту столбцов:

import org.apache.spark.sql.functions.{
  array, col, expr, lit, map_from_arrays, map_from_entries
}

val cols = Seq("a", "b", "c")

val dfm = df.withColumn(
  "cmap", 
  map_from_arrays(array(cols map lit: _*), array(cols map col: _*))
)

и transform the config:

dfm.withColumn(
  "config_mapped",
   map_from_entries(expr("transform(config, k -> struct(k, cmap[k]))"))
).show

// +-----+---+---+---+------+--------------------+----------------+
// |group|  a|  b|  c|config|                cmap|   config_mapped|
// +-----+---+---+---+------+--------------------+----------------+
// |    a|  1|  2|  3|   [a]|[a -> 1, b -> 2, ...|        [a -> 1]|
// |    b|  2|  3|  4|[a, b]|[a -> 2, b -> 3, ...|[a -> 2, b -> 3]|
// +-----+---+---+---+------+--------------------+----------------+