PySpark:具有不同列的 DataFrames 的动态联合

Posted

技术标签:

【中文标题】PySpark:具有不同列的 DataFrames 的动态联合【英文标题】:PySpark: dynamic union of DataFrames with different columns 【发布时间】:2018-11-06 04:38:56 【问题描述】:

考虑此处显示的数组。我有 3 组数组:

数组 1:

C1  C2  C3
1   2   3
9   5   6

数组 2:

C2 C3 C4
11 12 13
10 15 16

数组 3:

C1   C4
111  112
110  115

我需要如下输出,输入我可以获得 C1、...、C4 的任何一个值,但在加入时我需要获得正确的值,如果该值不存在,那么它应该为零。

预期输出:

C1 C2 C3 C4
1  2  3  0
9  5  6  0
0  11 12 13
0 10 15 16
111 0 0 112
110 0 0 115

我已经编写了 pyspark 代码,但是我已经硬编码了新列的值及其 RAW,我需要将下面的代码转换为方法重载,以便我可以将此脚本用作自动脚本。我只需要使用 python/pyspark 而不是 pandas。

import pyspark
from pyspark import SparkContext
from pyspark.sql.functions import lit
from pyspark.sql import SparkSession

sqlContext = pyspark.SQLContext(pyspark.SparkContext())

df01 = sqlContext.createDataFrame([(1, 2, 3), (9, 5, 6)], ("C1", "C2", "C3"))
df02 = sqlContext.createDataFrame([(11,12, 13), (10, 15, 16)], ("C2", "C3", "C4"))
df03 = sqlContext.createDataFrame([(111,112), (110, 115)], ("C1", "C4"))

df01_add = df01.withColumn("C4", lit(0)).select("c1","c2","c3","c4")
df02_add = df02.withColumn("C1", lit(0)).select("c1","c2","c3","c4")
df03_add = df03.withColumn("C2", lit(0)).withColumn("C3", lit(0)).select("c1","c2","c3","c4")

df_uni = df01_add.union(df02_add).union(df03_add)
df_uni.show()

方法重载示例:

class Student:
     def ___Init__ (self,m1,m2):
         self.m1 = m1
         self.m2 = m2

     def sum(self,c1=None,c2=None,c3=None,c4=None):
         s = 0
         if c1!= None and c2 != None and c3 != None:
            s = c1+c2+c3
         elif c1 != None and c2 != None:
             s = c1+c2
         else:
            s = c1
         return s

print(s1.sum(55,65,23))

【问题讨论】:

【参考方案1】:

可能有很多更好的方法可以做到这一点,但也许下面的方法对将来的任何人都有用。

from pyspark.sql import SparkSession
from pyspark.sql.functions import lit

spark = SparkSession.builder\
    .appName("DynamicFrame")\
    .getOrCreate()

df01 = spark.createDataFrame([(1, 2, 3), (9, 5, 6)], ("C1", "C2", "C3"))
df02 = spark.createDataFrame([(11,12, 13), (10, 15, 16)], ("C2", "C3", "C4"))
df03 = spark.createDataFrame([(111,112), (110, 115)], ("C1", "C4"))

dataframes = [df01, df02, df03]

# Create a list of all the column names and sort them
cols = set()
for df in dataframes:
    for x in df.columns:
        cols.add(x)
cols = sorted(cols)

# Create a dictionary with all the dataframes
dfs = 
for i, d in enumerate(dataframes):
    new_name = 'df' + str(i)  # New name for the key, the dataframe is the value
    dfs[new_name] = d
    # Loop through all column names. Add the missing columns to the dataframe (with value 0)
    for x in cols:
        if x not in d.columns:
            dfs[new_name] = dfs[new_name].withColumn(x, lit(0))
    dfs[new_name] = dfs[new_name].select(cols)  # Use 'select' to get the columns sorted

# Now put it al together with a loop (union)
result = dfs['df0']      # Take the first dataframe, add the others to it
dfs_to_add = dfs.keys()  # List of all the dataframes in the dictionary
dfs_to_add.remove('df0') # Remove the first one, because it is already in the result
for x in dfs_to_add:
    result = result.union(dfs[x])
result.show()

输出:

+---+---+---+---+
| C1| C2| C3| C4|
+---+---+---+---+
|  1|  2|  3|  0|
|  9|  5|  6|  0|
|  0| 11| 12| 13|
|  0| 10| 15| 16|
|111|  0|  0|112|
|110|  0|  0|115|
+---+---+---+---+

【讨论】:

你说的可能更好,但我还没有遇到过,所以感谢这个新颖的代码!你曾经在 Scala 中做过类似的事情吗?或者看到在 Scala 中执行此操作的代码? @JoshuaJames - 这是 Scala 中的版本 - ***.com/a/60702657/9445912【参考方案2】:

我会试试的

df = df1.join(df2, ['each', 'shared', 'col], how='full')

【讨论】:

【参考方案3】:

从 Spark 3.1.0 开始,您可以使用 unionByNameallowMissingColumns=True,这几乎完全符合您的要求。您需要 fillna 将空值替换为 0。

请注意,原始数据框中的任何空值也将替换为 0。

dfs = [df01, df02, df03]

res = dfs[0]
for df in dfs[1:]:
    res = res.unionByName(df, allowMissingColumns=True)

res.fillna(0).show()

+---+---+---+---+                                                               
| C1| C2| C3| C4|
+---+---+---+---+
|  1|  2|  3|  0|
|  9|  5|  6|  0|
|  0| 11| 12| 13|
|  0| 10| 15| 16|
|111|  0|  0|112|
|110|  0|  0|115|
+---+---+---+---+

【讨论】:

以上是关于PySpark:具有不同列的 DataFrames 的动态联合的主要内容,如果未能解决你的问题,请参考以下文章

PySpark DataFrames 是不是具有像 Pandas 中的“管道”功能?

Pyspark 从具有不同列的行/数据创建 DataFrame

PySpark DataFrames - 使用不同类型的列之间的比较进行过滤

PySpark DataFrames - 在不转换为 Pandas 的情况下进行枚举的方法?

PySpark DataFrames - 在不转换为 Pandas 的情况下进行枚举的方法?

是否可以使用 pyspark 过滤 Spark DataFrames 以返回列值在列表中的所有行?