PySpark：将两个数据框连接起来，同时对列求和

Question

我有两个PySpark数据框，希望进行左连接(left join)

Prev_table:
| user_id | earnings | start_date | end_date   |
|---------|--------|------------|------------|
| 1       | 10     | 2020-06-01 | 2020-06-10 |
| 2       | 20     | 2020-06-01 | 2020-06-10 |
| 3       | 30     | 2020-06-01 | 2020-06-10 |

New_table:
| user_id | profit |
|---------|--------|
| 1       | 100    |
| 2       | 200    |
| 5       | 500    |

合并后的表格是根据user_id进行分组的两个数据帧的连接结果。而earnings列则为Prev_table.earnings + New_table.profit。

| user_id | earnings | start_date | end_date   |
|---------|--------|------------|------------|
| 1       | 110    | 2020-06-01 | 2020-06-10 |
| 2       | 220    | 2020-06-01 | 2020-06-10 |
| 3       | 30     | 2020-06-01 | 2020-06-10 |
| 5       | 500    |            |            |

这就是 Pandas concat 的作用，但在 PySpark 中，我认为需要使用 PySparks union 来实现。 另外，我不确定如何对两列进行求和。 我知道需要使用类似于 combined_df.agg({"earnings": "sum"}).collect() 的东西，但我很难让它正常工作。 感谢您提供有关 PySpark 工作流程的任何指导。 谢谢。

- madu

Answer 1

可能有更好的方法，但一个方案是，将profit重命名为earnings，然后填充df2中缺失的列，然后使用union和agg进行分组聚合：

假设Prev_table是df1，New_table是df2

import pyspark.sql.functions as F

df3 = df2.select("user_id",F.col("profit").alias("earnings"))

(df1.union(df3.select("*",*[F.lit(None).alias(i) 
            for i in df1.columns if i not in df3.columns]))
.groupBy("user_id").agg(F.sum("earnings").alias("earnings")
 ,F.first("start_date",ignorenulls=True).alias("start_date")
 ,F.first("end_date",ignorenulls=True).alias("end_date")).orderBy("user_id")).show()

---

+-------+--------+----------+----------+
|user_id|earnings|start_date|  end_date|
+-------+--------+----------+----------+
|      1|     110|2020-06-01|2020-06-10|
|      2|     220|2020-06-01|2020-06-10|
|      3|      30|2020-06-01|2020-06-10|
|      5|     500|      null|      null|
+-------+--------+----------+----------+

- anky

Answer 2

其他解决方案的替代方案（使用scala编写，但只需进行最小更改即可在pyspark中使用）-

加载提供的输入

  val data1 =
      """
        |user_id | earnings | start_date | end_date
        |1       | 10     | 2020-06-01 | 2020-06-10
        |2       | 20     | 2020-06-01 | 2020-06-10
        |3       | 30     | 2020-06-01 | 2020-06-10
      """.stripMargin
    val stringDS1 = data1.split(System.lineSeparator())
      .map(_.split("\\|").map(_.replaceAll("""^[ \t]+|[ \t]+$""", "")).mkString(","))
      .toSeq.toDS()
    val df1 = spark.read
      .option("sep", ",")
      .option("inferSchema", "true")
      .option("header", "true")
      .option("nullValue", "null")
      .csv(stringDS1)
    df1.show(false)
    df1.printSchema()
    /**
      * +-------+--------+-------------------+-------------------+
      * |user_id|earnings|start_date         |end_date           |
      * +-------+--------+-------------------+-------------------+
      * |1      |10      |2020-06-01 00:00:00|2020-06-10 00:00:00|
      * |2      |20      |2020-06-01 00:00:00|2020-06-10 00:00:00|
      * |3      |30      |2020-06-01 00:00:00|2020-06-10 00:00:00|
      * +-------+--------+-------------------+-------------------+
      *
      * root
      * |-- user_id: integer (nullable = true)
      * |-- earnings: integer (nullable = true)
      * |-- start_date: timestamp (nullable = true)
      * |-- end_date: timestamp (nullable = true)
      */

    val data2 =
      """
        |user_id | profit
        |1       | 100
        |2       | 200
        |5       | 500
      """.stripMargin
    val stringDS2 = data2.split(System.lineSeparator())
      .map(_.split("\\|").map(_.replaceAll("""^[ \t]+|[ \t]+$""", "")).mkString(","))
      .toSeq.toDS()
    val df2 = spark.read
      .option("sep", ",")
      .option("inferSchema", "true")
      .option("header", "true")
      .option("nullValue", "null")
      .csv(stringDS2)
    df2.show(false)
    df2.printSchema()

    /**
      * +-------+------+
      * |user_id|profit|
      * +-------+------+
      * |1      |100   |
      * |2      |200   |
      * |5      |500   |
      * +-------+------+
      *
      * root
      * |-- user_id: integer (nullable = true)
      * |-- profit: integer (nullable = true)
      */

加入表并获取指定列

  df1.createOrReplaceTempView("prev_table")
    df2.createOrReplaceTempView("new_table")

   val processedDF = spark.sql(
      """
        |SELECT coalesce(p.user_id, n.user_id) as user_id,
        |       (coalesce(earnings,0) + coalesce(profit, 0)) as earnings,
        |        start_date,
        |        end_date
        |FROM prev_table p FULL OUTER JOIN new_table n ON p.user_id=n.user_id
      """.stripMargin)

     processedDF.orderBy("user_id").show(false)

    /**
      * +-------+--------+-------------------+-------------------+
      * |user_id|earnings|start_date         |end_date           |
      * +-------+--------+-------------------+-------------------+
      * |1      |110     |2020-06-01 00:00:00|2020-06-10 00:00:00|
      * |2      |220     |2020-06-01 00:00:00|2020-06-10 00:00:00|
      * |3      |30      |2020-06-01 00:00:00|2020-06-10 00:00:00|
      * |5      |500     |null               |null               |
      * +-------+--------+-------------------+-------------------+
      */

- Som

Answer 3

尝试使用连接操作，然后使用union all。

import pyspark.sql.functions as f

import pyspark.sql.functions as f

df = df1.join(df2, df1.user_id==df2.user_id, how='left').select(df1.user_id, (df1.earnings.cast('int')+f.when(df2.profit.cast('int').isNull(),0).otherwise(df2.profit)).alias('earnings'),df1.start_date,df1.end_date)
df3 = df2.join(df, df.user_id==df2.user_id, how='leftanti').select(df2.user_id,df2.profit.alias('earnings'),f.lit(None).alias('start_date'),f.lit(None).alias('end_date'))
final_df = df.union(df3)
final_df.show()

+-------+--------+----------+----------+
|user_id|earnings|start_date|  end_date|
+-------+--------+----------+----------+
|      3|    30.0|2020-06-01|2020-06-10|
|      1|   110.0|2020-06-01|2020-06-10|
|      2|   220.0|2020-06-01|2020-06-10|
|      5|     500|      null|      null|
+-------+--------+----------+----------+

- Shubham Jain

---

完全错过了 - Shubham Jain

尝试使用连接和联合，但你的方法看起来更干净。@anky - Shubham Jain