сплющивание вложенных json с использованием spark scala, создание 2 столбцов с одинаковыми именами и выдача ошибок дубликатов в Phoenix

Question

Я пытался сгладить очень вложенный JSON и создать искровой фрейм данных, и конечная цель - передать sh данный фрейм данных фениксу. Я успешно могу выровнять JSON, используя код.

def recurs(df: DataFrame): DataFrame = {
  if(df.schema.fields.find(_.dataType match {
    case ArrayType(StructType(_),_) | StructType(_) => true
    case _ => false
  }).isEmpty) df
  else {
    val columns = df.schema.fields.map(f => f.dataType match {
      case _: ArrayType => explode(col(f.name)).as(f.name)
      case s: StructType => col(s"${f.name}.*")
      case _ => col(f.name)
    })
    recurs(df.select(columns:_*))
  }
}
val df = spark.read.json(json_location)
flatten_df = recurs(df)
flatten_df.show()

Мой вложенный json выглядит примерно так:

          {
           "Total Value": 3,
           "Topic": "Example",
           "values": [
                      {
                        "value": "#example1",
                        "points": [
                                   [
                                   "123",
                                   "156"
                                  ]
                            ],
                        "properties": {
                         "date": "12-04-19",
                         "value": "Model example 1"
                            }
                         },
                       {"value": "#example2",
                        "points": [
                                   [
                                   "124",
                                   "157"
                                  ]
                            ],
                        "properties": {
                         "date": "12-05-19",
                         "value": "Model example 2"
                            }
                         }
                      ]
               }

Вывод, который я получаю:

+-----------+-----------+----------+-------------+------------------------+------------------------+
|Total Value| Topic     |value     | points      | date                   |    value               |
+-----------+-----------+----------+-------------+------------------------+------------------------+
| 3         | Example   | example1 | [123,156]   | 12-04-19               |    Model example 1     | 
| 3         | Example   | example2 | [124,157]   | 12-05-19               |    Model example 2     |       
+-----------+-----------+----------+-------------+------------------------+------------------------+

Итак, ключ значения встречается 2 раза в json, поэтому он создает имя 2 столбцов, но это ошибка, и в Phoenix эта ошибка не допускается для приема этих данных.

Сообщение об ошибке:

ОШИБКА 514 (42892): в определении объекта или операторе ALTER TABLE / VIEW обнаружено дублированное имя столбца

Я ожидаю этот вывод, чтобы Феникс мог дифференцироваться столбцы.

+-----------+-----------+--------------+---------------+------------------------+------------------------+
|Total Value| Topic     |values.value  | values.points | values.properties.date | values.properties.value|              |
+-----------+-----------+--------------+---------------+------------------------+------------------------+
| 3         | Example   | example1     | [123,156]     | 12-04-19               |    Model example 1     | 
| 3         | Example   | example2     | [124,157]     | 12-05-19               |    Model example 2     |       
+-----------+-----------+--------------+---------------+------------------------+------------------------+

Таким образом, Phoenix может идеально использовать данные, пожалуйста, предложите любые изменения в коде выравнивания или любую помощь для достижения этого. Спасибо

Answer 1

Вам необходимо внести небольшие изменения в метод recurs:

1. Работа с ArrayType(st: StructType, _) вместо ArrayType.
2. Избегайте использования * и назовите каждое поле в второе совпадение (StructType).
3. Используйте backticks в нужных местах, чтобы переименовать поля с сохранением именования приоритетов.

Вот некоторый код:

def recurs(df: DataFrame): DataFrame = {
  if(!df.schema.fields.exists(_.dataType match {
    case ArrayType(StructType(_),_) | StructType(_) => true
    case _ => false
  })) df
  else {
    val columns = df.schema.fields.flatMap(f => f.dataType match {
      case ArrayType(st: StructType, _) => Seq(explode(col(f.name)).as(f.name))
      case s: StructType =>
        s.fieldNames.map{sf => col(s"`${f.name}`.$sf").as(s"${f.name}.$sf")}
      case _ => Seq(col(s"`${f.name}`"))
    })
    recurs(df.select(columns:_*))
  }
}

val newDF = recurs(df).cache
newDF.show(false)
newDF.printSchema

И новый вывод:

+-------+-----------+-------------+----------------------+-----------------------+------------+
|Topic  |Total Value|values.points|values.properties.date|values.properties.value|values.value|
+-------+-----------+-------------+----------------------+-----------------------+------------+
|Example|3          |[[123, 156]] |12-04-19              |Model example 1        |#example1   |
|Example|3          |[[124, 157]] |12-05-19              |Model example 2        |#example2   |
+-------+-----------+-------------+----------------------+-----------------------+------------+

root
 |-- Topic: string (nullable = true)
 |-- Total Value: long (nullable = true)
 |-- values.points: array (nullable = true)
 |    |-- element: array (containsNull = true)
 |    |    |-- element: string (containsNull = true)
 |-- values.properties.date: string (nullable = true)
 |-- values.properties.value: string (nullable = true)
 |-- values.value: string (nullable = true)