合并列是变量结构的数据框 - Pyspark

Posted 2023-04-15

【中文标题】合并列是变量结构的数据框 - Pyspark

【英文标题】：Merging a dataframes where a column is a variable struct - Pyspark

【发布时间】：2020-04-02 11:05:17

【问题描述】：

我有一堆需要合并的数据框，它们拥有的 4 列是相同的，但其中的一列（参数）根据数据框的不同有可​​变字段，我在下面显示了示例：

+---------+-----------+--------------------------------------------------------------------------------------------------------------------------+-------------------+
|attribute|operation  |params                                                                                                                    |timestamp          |
+---------+-----------+--------------------------------------------------------------------------------------------------------------------------+-------------------+
|profile  |CREATE_CARD|[50d966f2-2820-441a-afbe-851e45eeb13e, 1s9miu7t6an50fplvvhybow6edx9_STG, 993270335, CREATED_CARD, 8236961209881953, kobo] |2020-02-24 03:07:04|
+---------+-----------+--------------------------------------------------------------------------------------------------------------------------+-------------------+

+---------+---------+--------------------------------------------------------------------------------------------------+-------------------+

|attribute|operation|params                                                                                            |timestamp          |
+---------+---------+--------------------------------------------------------------------------------------------------+-------------------+
|profile  |UPDATE   |[0792b8d1-7ad9-43fc-9e75-9b1f2612834c, rkm9a7mescuwp0s4i01zlwi2ftu9_STG, 993270329, primary_email]|2020-02-12 18:13:08|
+---------+---------+--------------------------------------------------------------------------------------------------+-------------------+

+---------+---------+-----------------------------------------------------------------------------------+-------------------+
|attribute|operation|params                                                                             |timestamp          |
+---------+---------+-----------------------------------------------------------------------------------+-------------------+
|member   |CREATE   |[ea8e7e39-4a0a-4d41-b47e-70c8e56a2bca, h4m015wf1qxwrogj6d9l2uc5bsa9_STG, 993270331]|2020-01-02 09:51:32|
+---------+---------+-----------------------------------------------------------------------------------+-------------------+

如何在不向缺失字段添加空值的情况下将这些数据帧中的所有行放入单个数据帧？ 我必须合并数据帧以存储按字段时间戳排序的最终数据帧。 我不想将参数保存为字符串，因为我需要将最终合并的数据帧作为 JSON 存储在文本中，并将其保存为字符串会将转义字符添加到最终文件中，这是我试图避免的。

我尝试使用 toJSON() 将 Dataframes 转换为 JSON 对象，然后将其合并，但 toJSON() 给了我一个带有字符串类型元素的 RDD，我无法对其进行排序。 我也尝试过union，但这不起作用，因为在上面显示的每个数据框中，“params”列是不同的结构。 最有效的方法是什么？

最终输出应如下所示：

+---------+-----------+--------------------+-------------------------------------------------------------------------------------------------------------------------+
|attribute|operation  |timestamp           |params                                                    
+---------+-----------+--------------------+-------------------------------------------------------------------------------------------------------------------------+
|profile  |CREATE_CARD|2020-02-24 03:07:04 |[50d966f2-2820-441a-afbe-851e45eeb13e, 1s9miu7t6an50fplvvhybow6edx9_STG, 993270335, CREATED_CARD, 8236961209881953, kobo]|
|profile  |UPDATE     |2020-02-12 18:13:08 |[0792b8d1-7ad9-43fc-9e75-9b1f2612834c, rkm9a7mescuwp0s4i01zlwi2ftu9_STG, 993270329, primary_email]|
|member   |CREATE     |2020-01-02 09:51:32 |[ea8e7e39-4a0a-4d41-b47e-70c8e56a2bca, h4m015wf1qxwrogj6d9l2uc5bsa9_STG, 993270331]

【问题讨论】：

合并这些数据帧 -- 并不完全清楚所需的输出应该是什么样子。

【参考方案1】：

您可以使用“unionByName”（从 2.3 开始）https://spark.apache.org/docs/latest/api/python/_modules/pyspark/sql/dataframe.html

喜欢：

from functools import reduce
from pyspark.sql import DataFrame

dfs = [df1, df2, df3]
df = reduce(DataFrame.unionByName, dfs)

如果您的 Spark 版本低于 2.3，您可以使用“联合”，但要注意列顺序。

【讨论】：

我试过这个，得到一个错误说Union can only be performed on tables with the compatible column types，因为列参数是每个数据帧中的不同结构

啊，是的：/ 一种解决方案可能是使用explode，像上面一样使用unionByName，最后使用dfs.groupBy("attribute, "operation", "timestamp").agg(collect_list("col").as("params")) 可能远非最优，取决于分组条件是否唯一，但是应该做的工作。