Транспонировать DataFrame одну строку в столбец в Spark с scala

Question

Я видел этот вопрос здесь: Транспонировать DataFrame без агрегирования в Spark с scala, и я хотел сделать прямо противоположное.

У меня есть этот Dataframe с одной строкой, с значения типа string, int, bool, array:

+-----+-------+-----+------+-----+
|col1 | col2  |col3 | col4 |col5 |
+-----+-------+-----+------+-----+
|val1 | val2  |val3 | val4 |val5 |
+-----+-------+-----+------+-----+

И я хочу транспонировать это так:

+-----------+-------+
|Columns    | values|
+-----------+-------+
|col1       | val1  |
|col2       | val2  |
|col3       | val3  |
|col4       | val4  |
|col5       | val5  |
+-----------+-------+

Я использую Apache Spark 2.4.3 с Scala 2.11

Редактировать: значения могут быть любого типа (int, double, bool, array), а не только строки.

Answer 1

Думал иначе, используя arrays_zip (который доступен в => Spark 2.4)] и получил следующее ...

Это будет работать для Spark => 2.0 и далее более простым способом (функции flatmap, map и explode) ...

Здесь функция map (используется со столбцом) создает новый столбец карты. Столбцы ввода должны быть сгруппированы как пары ключ-значение.

Случай: строковый тип данных в данных:

import org.apache.spark.sql.functions._

val df: DataFrame =Seq((("val1"),("val2"),("val3"),("val4"),("val5"))).toDF("col1","col2","col3","col4","col5")

var columnsAndValues = df.columns.flatMap { c => Array(lit(c), col(c)) }
df.printSchema()

df.withColumn("myMap", map(columnsAndValues:_*)).select(explode($"myMap"))
.toDF("Columns","Values").show(false)

Результат:

root
 |-- col1: string (nullable = true)
 |-- col2: string (nullable = true)
 |-- col3: string (nullable = true)
 |-- col4: string (nullable = true)
 |-- col5: string (nullable = true)


+-------+------+
|Columns|Values|
+-------+------+
|col1   |val1  |
|col2   |val2  |
|col3   |val3  |
|col4   |val4  |
|col5   |val5  |
+-------+------+

Случай: сочетание типы данных в Data:

Если у вас есть разные типы, преобразуйте их в String ... остальные шаги не изменятся ..

val df1 = df.select(df.columns.map(c => col(c).cast(StringType)): _*)

Полный пример:

import org.apache.spark.sql.functions._
  import spark.implicits._

  val df = Seq(((2), (3), (true), (2.4), ("val"))).toDF("col1", "col2", "col3", "col4", "col5")
  df.printSchema()
  /**
    * convert all columns to  to string type since its needed further
    */
  val df1 = df.select(df.columns.map(c => col(c).cast(StringType)): _*)
  df1.printSchema()
  var ColumnsAndValues: Array[Column] = df.columns.flatMap { c => {
    Array(lit(c), col(c))
  }
  }

  df1.withColumn("myMap", map(ColumnsAndValues: _*))
    .select(explode($"myMap"))
    .toDF("Columns", "Values")
    .show(false)

Результат:

root
 |-- col1: integer (nullable = false)
 |-- col2: integer (nullable = false)
 |-- col3: boolean (nullable = false)
 |-- col4: double (nullable = false)
 |-- col5: string (nullable = true)

root
 |-- col1: string (nullable = false)
 |-- col2: string (nullable = false)
 |-- col3: string (nullable = false)
 |-- col4: string (nullable = false)
 |-- col5: string (nullable = true)

+-------+------+
|Columns|Values|
+-------+------+
|col1   |2     |
|col2   |3     |
|col3   |true  |
|col4   |2.4   |
|col5   |val   |
+-------+------+

Answer 2

С Spark-2.4 Используйте arrays_zip с array(column_values), array(column_names), затем взорвитесь, чтобы получить результат.

Example:

val df=Seq((("val1"),("val2"),("val3"),("val4"),("val5"))).toDF("col1","col2","col3","col4","col5")

val cols=df.columns.map(x => col(s"${x}"))

val str_cols=df.columns.mkString(",")

df.withColumn("new",explode(arrays_zip(array(cols:_*),split(lit(str_cols),",")))).
select("new.*").
toDF("values","Columns").
show()
//+------+-------+
//|values|Columns|
//+------+-------+
//|  val1|   col1|
//|  val2|   col2|
//|  val3|   col3|
//|  val4|   col4|
//|  val5|   col5|
//+------+-------+

---

UPDATE:

val df=Seq(((2),(3),(true),(2.4),("val"))).toDF("col1","col2","col3","col4","col5")

df.printSchema
//root
// |-- col1: integer (nullable = false)
// |-- col2: integer (nullable = false)
// |-- col3: boolean (nullable = false)
// |-- col4: double (nullable = false)
// |-- col5: string (nullable = true)

//cast to string
val cols=df.columns.map(x => col(s"${x}").cast("string").alias(s"${x}"))

val str_cols=df.columns.mkString(",")

df.withColumn("new",explode(arrays_zip(array(cols:_*),split(lit(str_cols),",")))).
select("new.*").
toDF("values","Columns").
show()

//+------+-------+
//|values|Columns|
//+------+-------+
//|     2|   col1|
//|     3|   col2|
//|  true|   col3|
//|   2.4|   col4|
//|   val|   col5|
//+------+-------+