展平嵌套数据框并重新转换为原始形式

Posted 2023-04-15

【中文标题】展平嵌套数据框并重新转换为原始形式

【英文标题】：Flatten nested dataframe and reconvert to original form

【发布时间】：2021-04-15 07:13:40

【问题描述】：

我有一个具有以下架构的镶木地板文件

root
 |-- listOfMetrics: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- Action: string (nullable = true)
 |    |    |-- id: long (nullable = true)
 |    |    |-- date: date (nullable = true)
 |    |    |-- female_executives: double (nullable = true)
 |    |    |-- male_executives: double (nullable = true)
 |    |    |-- female_directors: double (nullable = true)
 |    |    |-- male_directors: double (nullable = true)
 |    |    |-- female_executives_and_directors: double (nullable = true)
 |    |    |-- male_executives_and_directors: double (nullable = true)
 |    |    |-- flag: integer (nullable = true)

并且 df.show() 返回类似于以下内容的内容

+----------------------------------+
|                    listOfMetrics |
+----------------------------------+
|                 [[ADD, 5394, 2...|
|                 [[ADD, 527, 20...|
|                 [[ADD, 714, 20...|
|                 [[ADD, 765, 20...|
|                 [[ADD, 996, 20...|
|                 [[ADD, 146, 20...|
|                 [[ADD, 947, 20...|
+----------------------------------+

“操作”列是我的目标。此列可以包含“删除”或“添加”，因此基于此，我需要分隔行。 我采用的方法是展平，然后使用 pyspark.sql 分离并重新转换回其原始形式，但在转换步骤失败。 我有以下问题

那么在 pyspark 中是否有更好的方法来做到这一点？ 我们能否在转换后将数据帧动态转换为其原始形式？ 我们能否在不展平行的情况下分隔行？

我是 spark 新手，发现很难让它发挥作用

谢谢

【参考方案1】：

可以使用filter 将每个数组分成两部分：一部分仅包含ADD 的元素，另一部分仅包含DELETE 的元素。之后，每个部分都将成为使用stack 的单独行。可以保留原始结构，不需要展平数据框。

我正在使用一组稍微简化的测试数据：

root
 |-- listOfMetrics: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- Action: string (nullable = true)
 |    |    |-- flag: long (nullable = true)
 |    |    |-- id: long (nullable = true)

+----------------------------------------------------------------------------------+
|listOfMetrics                                                                     |
+----------------------------------------------------------------------------------+
|[[ADD, 2, 1], [ADD, 4, 3], [DELETE, 6, 5], [DELETE, 8, 7], [ADD, 10, 9]]          |
|[[ADD, 12, 11], [ADD, 14, 13], [DELETE, 16, 15], [DELETE, 18, 17], [ADD, 110, 19]]|
+----------------------------------------------------------------------------------+

代码：

df.withColumn("listOfMetrics", F.expr("""stack(2, 
            filter(listOfMetrics, e -> e.Action = 'ADD'), 
            filter(listOfMetrics, e -> e.Action = 'DELETE'))""")) \
    .show(truncate=False)

输出：

+----------------------------------------------+
|listOfMetrics                                 |
+----------------------------------------------+
|[[ADD, 2, 1], [ADD, 4, 3], [ADD, 10, 9]]      |
|[[DELETE, 6, 5], [DELETE, 8, 7]]              |
|[[ADD, 12, 11], [ADD, 14, 13], [ADD, 110, 19]]|
|[[DELETE, 16, 15], [DELETE, 18, 17]]          |
+----------------------------------------------+