Преобразование столбцов таблицы и значений во вложенные JSON

Question

У меня есть CSV-файл со столбцами, как показано ниже

Требуемый результат для вложенных JSON объектов, как показано ниже:

{
"Type" : "A"
"Value" :"1"

}

{
"Type" : "B"
"Value" :"1"

}

Я попробовал следующий код: любая помощь будет высоко оценена.

from pyspark.sql.functions import *

    list1=['A','B','C']
    df2=spark.sql("select * from test   limit  10" )
    df3=df2.select(list1)
    for i in range(0,len(list1)):
      df4=df3.withColumn("Type",lit(list1[i]))

Answer 1

Это можно сделать с помощью трюка UnPivoting ...

Предположим, у вас есть набор данных, как показано ниже. Давайте назовем его результатами теста пациента. Столбец A, B, C означает ... Тип теста A, Тип теста B, ... и значения в этих столбцах означают результаты теста в цифрах c

+-------------+---+---+---+---+---+---+---+
|PatientNumber|  A|  B|  C|  D|  E|  F|  G|
+-------------+---+---+---+---+---+---+---+
|          101|  1|  2|  3|  4|  5|  6|  7|
|          102| 11| 12| 13| 14| 15| 16| 17|
+-------------+---+---+---+---+---+---+---+

Я добавил столбец PatientNumber только для того, чтобы данные выглядели более осмысленными. Вы можете удалить это из своего кода.

Я добавил этот набор данных в CSV ..

val testDF = spark.read.format("csv").option("header", "true").load("""C:\TestData\CSVtoJSon.csv""")

Давайте создадим 2 массива, один для столбцов идентификаторов, а другой для всех типов тестов.

val idCols = Array("PatientNumber")

val valCols = testDF.columns.diff(idCols)

Тогда вот код Unpivot

val valcolNames = valCols.map(x => List(''' + x + ''', x))
val unPivotedDF = testDF.select($"PatientNumber", expr(s"""stack(${valCols.size},${valcolNames.flatMap(x => x).mkString(",")} ) as (Type,Value)"""))

Вот как выглядят эти нерасщепленные данные -

+-------------+----+-----+
|PatientNumber|Type|Value|
+-------------+----+-----+
|          101|   A|    1|
|          101|   B|    2|
|          101|   C|    3|
|          101|   D|    4|
|          101|   E|    5|
|          101|   F|    6|
|          101|   G|    7|
|          102|   A|   11|
|          102|   B|   12|
|          102|   C|   13|
|          102|   D|   14|
|          102|   E|   15|
|          102|   F|   16|
|          102|   G|   17|
+-------------+----+-----+

Наконец, запишите этот Unpivoted DF как Json -

unPivotedDF.coalesce(1).write.format("json").mode("Overwrite").save("""C:\TestData\output""")

Содержимое Json Файл выглядит так же, как ваш желаемый результат -

{"PatientNumber":"101","Type":"A","Value":"1"}
{"PatientNumber":"101","Type":"B","Value":"2"}
{"PatientNumber":"101","Type":"C","Value":"3"}
{"PatientNumber":"101","Type":"D","Value":"4"}
{"PatientNumber":"101","Type":"E","Value":"5"}
{"PatientNumber":"101","Type":"F","Value":"6"}
{"PatientNumber":"101","Type":"G","Value":"7"}
{"PatientNumber":"102","Type":"A","Value":"11"}
{"PatientNumber":"102","Type":"B","Value":"12"}
{"PatientNumber":"102","Type":"C","Value":"13"}
{"PatientNumber":"102","Type":"D","Value":"14"}
{"PatientNumber":"102","Type":"E","Value":"15"}
{"PatientNumber":"102","Type":"F","Value":"16"}
{"PatientNumber":"102","Type":"G","Value":"17"}