将元素添加到结构数组并合并Spark 2.3中的结构数组

Question

我有下面的数据帧模式作为df.current模式，需要获得预期的模式作为df.expected模式，有没有一种方法，我可以在火花2.3实现这一点

df.current架构：

 |-- enqueuedTime: timestamp (nullable = true)
 |-- VIN: string (nullable = true)
 |-- TT: long (nullable = true)
 |-- MSG_TYPE: string (nullable = true)
 |-- ADA: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- E: long (nullable = true)
 |    |    |-- V: double (nullable = true)
 |    |    |-- E: long (nullable = true)
 |    |    |-- V: double (nullable = true)
 |-- ADW: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- E: long (nullable = true)
 |    |    |-- V: double (nullable = true)

df。预期架构：

 |-- enqueuedTime: timestamp (nullable = true)
 |-- VIN: string (nullable = true)
 |-- TT: long (nullable = true)
 |-- MSG_TYPE: string (nullable = true)
 |-- SIGNAL: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- SN: string (nullable = true)
 |    |    |-- E:  long (nullable = true)
 |    |    |-- V:  double (nullable = true)
 |    |    |-- SN: string (nullable = true) 
 |    |    |-- E:  long (nullable = true)
 |    |    |-- V:  double (nullable = true)
 |    |    |-- SN: string (nullable = true)
 |    |    |-- E: long (nullable = true)
 |    |    |-- V: double (nullable = true)

示例数据：

 +----------------+---+---------+-----------------------------------------------------+--------------------------+
|vin              |tt |msg_type |ada                                                  |adw                       |                      |
+-----------------+---+---------+-----------------------------------------------------+--------------------------+
|FU7XXXXXXXXXXXXXX|0  |SIGNAL   |[{"E":15XXXXXXXX,"V":2, {"E":15XXXXXXXX,"V":1}]      |null                      |                          
|FU7XXXXXXXXXXXXXX|0  |SIGNAL   |null                                                 |[{"E":15XXXXXXXX,"V":3}]  |                      
|FU7XXXXXXXXXXXXXX|0  |SIGNAL   |null                                                 |[{"E":15XXXXXXXX,"V":4.1}]|
+-----------------+---+---------+--------------------------+--------------------------+--------------------------+

注意：这里需要实现两件事：

1. 为元素中的每个E、V对创建新字段SN，其值应为数组名称。例如：对于第一个数组列（ADA），SN的值=ADA
2. 将阵列（ADA、ADW）合并为一个外部阵列（信号）

Answer 1

您正在查找的模式不正确

scala> newDF.printSchema
root
 |-- ADA: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- E: string (nullable = true)
 |    |    |-- V: long (nullable = true)
 |-- ADW: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- E: string (nullable = true)
 |    |    |-- V: long (nullable = true)
 |-- MSG_TYPE: string (nullable = true)
 |-- number: long (nullable = true)
 |-- tt: long (nullable = true)
 |-- vin: string (nullable = true)
 |-- sig: struct (nullable = false)
 |    |-- SN: array (nullable = true)
 |    |    |-- element: struct (containsNull = true)
 |    |    |    |-- E: string (nullable = true)
 |    |    |    |-- V: long (nullable = true)
 |    |-- SN: array (nullable = true)
 |    |    |-- element: struct (containsNull = true)
 |    |    |    |-- E: string (nullable = true)
 |    |    |    |-- V: long (nullable = true)

如果您对这个模式很满意，请在实现它的过程中进一步阅读。

创建虚拟数据以复制您的模式（此步骤可以忽略）

scala> val vas = """{"df":[ { "vin": "FU7XXXXXXXXXXXXXX", "tt": 0, "MSG_TYPE": "SIGNAL", "number": 123, "ADA": [{"E":"15XXXXXXXX","V":2}, {"E":"15XXXXXXXX","V":1}], "ADW": [{"E":"15XXXXXXXX","V":2}, {"E":"15XXXXXXXX","V":1}] }, { "vin": "FU7XXXXXXXXXXXXXX", "tt": 0, "MSG_TYPE": "SIGNAL", "number": 123, "ADA": [{"E":"15XXXXXXXX","V":2}, {"E":"15XXXXXXXX","V":1}], "ADW": [{"E":"15XXXXXXXX","V":2}, {"E":"15XXXXXXXX","V":1}] }, { "vin": "FU7XXXXXXXXXXXXXX", "tt": 0, "MSG_TYPE": "SIGNAL", "number": 123, "ADA": [{"E":"15XXXXXXXX","V":2}, {"E":"15XXXXXXXX","V":1}], "ADW":[{"E":"15XXXXXXXX","V":2}, {"E":"15XXXXXXXX","V":1}] }] }"""
vas: String = {"df":[ { "vin": "FU7XXXXXXXXXXXXXX", "tt": 0, "MSG_TYPE": "SIGNAL", "number": 123, "ADA": [{"E":"15XXXXXXXX","V":2}, {"E":"15XXXXXXXX","V":1}], "ADW": [{"E":"15XXXXXXXX","V":2}, {"E":"15XXXXXXXX","V":1}] }, { "vin": "FU7XXXXXXXXXXXXXX", "tt": 0, "MSG_TYPE": "SIGNAL", "number": 123, "ADA": [{"E":"15XXXXXXXX","V":2}, {"E":"15XXXXXXXX","V":1}], "ADW": [{"E":"15XXXXXXXX","V":2}, {"E":"15XXXXXXXX","V":1}] }, { "vin": "FU7XXXXXXXXXXXXXX", "tt": 0, "MSG_TYPE": "SIGNAL", "number": 123, "ADA": [{"E":"15XXXXXXXX","V":2}, {"E":"15XXXXXXXX","V":1}], "ADW":[{"E":"15XXXXXXXX","V":2}, {"E":"15XXXXXXXX","V":1}] }] }

scala> val df = spark.read.json(Seq(vas).toDS).toDF.withColumn("arr", explode($"df")).select("arr.*")
df: org.apache.spark.sql.DataFrame = [ADA: array<struct<E:string,V:bigint>>, ADW: array<struct<E:string,V:bigint>> ... 4 more fields]

我希望您的数据是这样的：

scala> df.show(false)
+--------------------------------+--------------------------------+--------+------+---+-----------------+
|ADA                             |ADW                             |MSG_TYPE|number|tt |vin              |
+--------------------------------+--------------------------------+--------+------+---+-----------------+
|[[15XXXXXXXX,2], [15XXXXXXXX,1]]|[[15XXXXXXXX,2], [15XXXXXXXX,1]]|SIGNAL  |123   |0  |FU7XXXXXXXXXXXXXX|
|[[15XXXXXXXX,2], [15XXXXXXXX,1]]|[[15XXXXXXXX,2], [15XXXXXXXX,1]]|SIGNAL  |123   |0  |FU7XXXXXXXXXXXXXX|
|[[15XXXXXXXX,2], [15XXXXXXXX,1]]|[[15XXXXXXXX,2], [15XXXXXXXX,1]]|SIGNAL  |123   |0  |FU7XXXXXXXXXXXXXX|
+--------------------------------+--------------------------------+--------+------+---+-----------------+

实现所需输出的步骤

scala> val newDF = df.withColumn("sig", struct($"ADA".as("SN"), $"ADW".as("SN")))
newDF: org.apache.spark.sql.DataFrame = [ADA: array<struct<E:string,V:bigint>>, ADW: array<struct<E:string,V:bigint>> ... 5 more fields]

scala> newDF.printSchema
root
 |-- ADA: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- E: string (nullable = true)
 |    |    |-- V: long (nullable = true)
 |-- ADW: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- E: string (nullable = true)
 |    |    |-- V: long (nullable = true)
 |-- MSG_TYPE: string (nullable = true)
 |-- number: long (nullable = true)
 |-- tt: long (nullable = true)
 |-- vin: string (nullable = true)
 |-- sig: struct (nullable = false)
 |    |-- SN: array (nullable = true)
 |    |    |-- element: struct (containsNull = true)
 |    |    |    |-- E: string (nullable = true)
 |    |    |    |-- V: long (nullable = true)
 |    |-- SN: array (nullable = true)
 |    |    |-- element: struct (containsNull = true)
 |    |    |    |-- E: string (nullable = true)
 |    |    |    |-- V: long (nullable = true)

我试着写这个数据框

newDF.write.mode("overwrite").parquet(path + "newDF.parquet")