在 PySpark 中有效地合并两个或多个数据帧/rdd答案

【问题标题】：Merging two or more dataframes/rdd efficiently in PySpark在 PySpark 中有效地合并两个或多个数据帧/rdd
【发布时间】：2021-02-09 21:45:37
【问题描述】：

我正在尝试基于相同的键合并三个 RDD。以下是数据。

+------+---------+-----+                                    
|UserID|UserLabel|Total|
+------+---------+-----+
|     2|    Panda|   15|
|     3|    Candy|   15|
|     1|  Bahroze|   15|
+------+---------+-----+
+------+---------+-----+
|UserID|UserLabel|Total|
+------+---------+-----+
|     2|    Panda| 7342|
|     3|    Candy| 5669|
|     1|  Bahroze| 8361|
+------+---------+-----+

+------+---------+-----+
|UserID|UserLabel|Total|
+------+---------+-----+
|     2|    Panda|   37|
|     3|    Candy|   27|
|     1|  Bahroze|   39|
+------+---------+-----+

我能够合并这三个 DF。我使用以下代码将它们转换为 RDD dict 用于所有三个

new_rdd = userTotalVisits.rdd.map(lambda row: row.asDict(True))

在 RDD 转换后，我将一个 RDD 和另外两个作为列表。映射第一个 RDD，然后根据相同的 UserID 添加其他键。我希望使用 pyspark 有更好的方法来做到这一点。这是我写的代码。

def transform(row):
    # Add a new key to each row
    for x in conversion_list: # first rdd in list of object as[{}] after using collect()
        if( x['UserID'] == row['UserID'] ):
            row["Total"] = { "Visitors": row["Total"], "Conversions": x["Total"]  }
    
    for y in Revenue_list: # second rdd in list of object as[{}] after using collect()
         if( y['UserID'] == row['UserID'] ):
            row["Total"]["Revenue"] = y["Total"]
    return row

potato = new_rdd.map(lambda row: transform(row)) #first rdd

我应该如何有效地合并这三个 RDD/DF？（因为我必须在一个巨大的 DF 上执行三个不同的任务）。寻找更有效的想法。 PS我还是火花新手。我的代码的结果如下，这是我需要的。

{'UserID': '2', 'UserLabel': 'Panda', 'Total': {'Visitors': 37, 'Conversions': 15, 'Revenue': 7342}}
{'UserID': '3', 'UserLabel': 'Candy', 'Total': {'Visitors': 27, 'Conversions': 15, 'Revenue': 5669}}
{'UserID': '1', 'UserLabel': 'Bahroze', 'Total': {'Visitors': 39, 'Conversions': 15, 'Revenue': 8361}}

谢谢。

【问题讨论】：

标签： apache-spark pyspark apache-spark-sql pyspark-dataframes

【解决方案1】：

您可以只对所有三个数据框进行左连接，但请确保您使用的第一个数据框具有所有 UserID 和 UserLabel 值。您可以按照@blackbishop 的建议忽略 GroupBy 操作，但它仍然会为您提供所需的输出

我正在展示如何在 scala 中完成，但您可以在 python 中执行类似的操作。

//source data
val visitorDF = Seq((2,"Panda",15),(3,"Candy",15),(1,"Bahroze",15),(4,"Test",25)).toDF("UserID","UserLabel","Total")
val conversionsDF = Seq((2,"Panda",37),(3,"Candy",27),(1,"Bahroze",39)).toDF("UserID","UserLabel","Total")
val revenueDF = Seq((2,"Panda",7342),(3,"Candy",5669),(1,"Bahroze",8361)).toDF("UserID","UserLabel","Total")

import org.apache.spark.sql.functions._
val finalDF = visitorDF.as("v").join(conversionsDF.as("c"),Seq("UserID","UserLabel"),"left")
.join(revenueDF.as("r"),Seq("UserID","UserLabel"),"left")
.withColumn("TotalArray",struct($"v.Total".as("Visitor"),$"c.Total".as("Conversions"),$"r.Total".as("Revenue")))
.drop("Total")
display(finalDF)

你可以看到如下输出：

【讨论】：

【解决方案2】：

您可以在 ["UserID", "UserLabel"] 列上加入 3 个数据框，从总共 3 个列中创建一个新结构 total：

from pyspark.sql import functions as F

result = df1.alias("conv") \
    .join(df2.alias("rev"), ["UserID", "UserLabel"], "left") \
    .join(df3.alias("visit"), ["UserID", "UserLabel"], "left") \
    .select(
    F.col("UserID"),
    F.col("UserLabel"),
    F.struct(
        F.col("conv.Total").alias("Conversions"),
        F.col("rev.Total").alias("Revenue"),
        F.col("visit.Total").alias("Visitors")
    ).alias("Total")
)

# write into json file
result.write.json("output")

# print result:
for i in result.toJSON().collect():
    print(i)

# {"UserID":3,"UserLabel":"Candy","Total":{"Conversions":15,"Revenue":5669,"Visitors":27}}
# {"UserID":1,"UserLabel":"Bahroze","Total":{"Conversions":15,"Revenue":8361,"Visitors":39}}
# {"UserID":2,"UserLabel":"Panda","Total":{"Conversions":15,"Revenue":7342,"Visitors":37}}

【讨论】：