python - 结合多个带有不同列的pyspark数据框

This question already has an answer here:

PySpark: dynamic union of DataFrames with different columns

                                    （1个答案）


                                上个月关闭。


我有一本包含20多个数据框的字典。它们每个都有不同的列和大小。

我想遍历字典并一次比较2个数据帧（left_df和rigt_df），将right_df中不存在的列添加到left_df，反之亦然。目标是合并所有20多个df，最后得到一个单个数据帧。

这是我尝试的方法，我首先将dfs分为2个字典：

import pyspark.sql.functions as F

d1 = {df1, df3, df5, df7, df9, df11, df13....}
d2 = {df2, df4, df6, df7, df8, df10, df12....}
for x in d1:
    left_df = x
    for y in d2:
        right_df = y
#add missing columns to x
for column in set(y.columns) - set(x.columns):
    left_df = left_df.withColumn(column, Flit(None))
#add missing columns to y
for column in set(x.columns) - set(y.columns):
    right_df = right_df.withColumn(column, Flit(None))
#order columns
allDfs = left_df.union(right_df.select(left_df.column))
allDfs.show()

我仅用2个数据帧尝试了此操作，其中每个字典中只有一个df，它确实给了我想要的结果，但是当我有2个以上时，它似乎并没有达到相同的效果。

最佳答案

将两个具有不同列的DataFrame合并起来很简单，只需将缺少的列添加为null，然后调用unionAll。

这是一个需要两个DataFrame并返回其并集的函数。
它假定两个DataFrame中都缺少列，但公共列必须具有相同的数据类型。

def union_dataframes(df1, df2):
    # use dict to preserve columns order, set does not
    cols1 = list(dict.fromkeys(df1.columns).keys())
    cols2 = list(dict.fromkeys(df2.columns).keys())
    merged_cols = list(dict.fromkeys(cols1 + cols2).keys())

    complete_cols1 = [col(c).alias(c) if c in cols1 else lit(None).alias(c) for c in merged_cols]
    complete_cols2 = [col(c).alias(c) if c in cols2 else lit(None).alias(c) for c in merged_cols]

    return df1.select(complete_cols1).unionAll(df2.select(complete_cols2))

然后，您可以将所有DataFrames放在列表中，并使用上述函数将其缩小：

list_df = [df1, df2, df3, ...] 
result_df = reduce(lambda a, b: union_dataframes(a, b), list_df)

关于python - 结合多个带有不同列的pyspark数据框，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/59505371/

上一篇：python - 删除 python 中 TFIDF 结果的同义词

下一篇：python - 列表项的assertEqual自定义比较

相关文章：

python - Python 2.6 中的可迭代枚举

performance - 在 Apache Spark 上迭代运行查询

python - 涉及 pyspark 的 CI/CD 测试 - 未设置 JAVA_HOME

join - 基于另一个没有显式连接列的 Spark df 计算 Spark df 中的新列

pyspark - pyspark udf 的参数数量可变

python - 我在设置 django-pagination 时遇到问题

python - Django-modpython 部署项目

Python - 函数返回值

hadoop - 配置单元分区列未通过sc.textFile在rdd中出现

apache-spark - 我可以重用参数实例来从 Spark 中的聚合函数返回值吗？