【发布时间】:2021-11-30 17:33:16
【问题描述】:
我有两张桌子:
表1:
| Row1 | Row2 | Common |
|---|---|---|
| x1 | y1 | c1 |
| x2 | y2 | c2 |
| x3 | y3 | c3 |
表2:
| Common | Value | Other |
|---|---|---|
| c1 | 5 | p1 |
| c1 | 6 | p2 |
| c1 | 10 | p3 |
| c2 | 22 | p4 |
| c2 | 14 | p5 |
| c3 | 6 | p6 |
| c3 | 21 | p7 |
| c3 | 11 | p8 |
需要根据这些连接这两个表:
首先需要使用公共列的 max(Value) 组从 Table2 中创建一个临时表,即:
| Common | Value | Other |
|---|---|---|
| c1 | 10 | p3 |
| c2 | 22 | p4 |
| c3 | 21 | p7 |
然后我需要将这个临时表与 Table1 加入即:
| Row1 | Row2 | Common | Value | Other |
|---|---|---|---|---|
| x1 | y1 | c1 | 10 | p3 |
| x2 | y2 | c2 | 22 | p4 |
| x3 | y3 | c3 | 21 | p7 |
现在,我需要在 PySpark 上实现它。
我用过这个sn-p:
import pyspark.sql.functions as f
Table1.alias("T1").join(
Table2.alias("T2"),
(col("T1.Common") == col("T2.Common") & col("T2.Value") == T2.groupBy('Common').agg(f.max('Value'))),
"left",
)
但它失败了。
我应该做些什么改变?
基本上,我需要编写这个 sql 查询的代码来制作那个临时表:
SELECT *
FROM Table2
WHERE Value IN(
SELECT max(Value)
FROM Table2
GROUP BY Common);
【问题讨论】:
-
先为 Table2 创建一个临时 df,然后将该 df 与 Table1 连接会更好吗?
-
是否需要为最大
Value的行保留Other列的值?此外,您似乎应该使用F.col而不是col,除非它已在其他地方导入。 -
是的,我需要保留其他列的值。
-
Value栏可以有联系吗?多行最大值相同怎么办? -
在表 2 中,只有一个“通用”数据(即 C1 或 C2)的“值”列将没有关联。因此,“其他”列中的两个数据之间也没有联系。
标签: sql python-3.x pyspark