我有以下pyspark df:
+------------------+--------+-------+
| ID| Assets|Revenue|
+------------------+--------+-------+
|201542399349300619| 1633944| 32850|
|201542399349300629| 3979760| 850914|
|201542399349300634| 3402687|1983568|
|201542399349300724| 1138291|1097553|
|201522369349300122| 1401406|1010828|
|201522369349300137| 16948| 171534|
|201522369349300142|13474056|2285323|
|201522369349300202| 481045| 241788|
|201522369349300207| 700861|1185640|
|201522369349300227| 178479| 267976|
+------------------+--------+-------+
对于每一行,我希望能够获得在Assets金额的20%范围内的行。例如,对于第一行(ID = 201542399349300619),我希望能够获得资产在20%+/- 1,633,944之间的所有行(因此在1,307,155到1,960,732之间):
+------------------+--------+-------+
| ID| Assets|Revenue|
+------------------+--------+-------+
|201542399349300619| 1633944| 32850|
|201522369349300122| 1401406|1010828|
使用此子集化表,我想获得平均资产并将其添加为新列。因此,对于上面的例子,它将是(1633944 + 1401406)= 1517675的平均资产
+------------------+--------+-------+---------+
| ID| Assets|Revenue|AvgAssets|
+------------------+--------+-------+---------+
|201542399349300619| 1633944| 32850| 1517675|
假设您的DataFrame具有类似于以下的模式(即Assets
和Revenue
是数字):
df.printSchema()
#root
# |-- ID: long (nullable = true)
# |-- Assets: integer (nullable = true)
# |-- Revenue: integer (nullable = true)
您可以在您提出的条件下将DataFrame发送到自己的join。在连接之后,您可以通过获取Assets
列的平均值进行分组和聚合。
例如:
from pyspark.sql.functions import avg, expr
df.alias("l")\
.join(
df.alias("r"),
on=expr("r.assets between l.assets*0.8 and l.assets*1.2")
)\
.groupBy("l.ID", "l.Assets", "l.Revenue")\
.agg(avg("r.Assets").alias("AvgAssets"))\
.show()
#+------------------+--------+-------+------------------+
#| ID| Assets|Revenue| AvgAssets|
#+------------------+--------+-------+------------------+
#|201542399349300629| 3979760| 850914| 3691223.5|
#|201522369349300202| 481045| 241788| 481045.0|
#|201522369349300207| 700861|1185640| 700861.0|
#|201522369349300137| 16948| 171534| 16948.0|
#|201522369349300142|13474056|2285323| 1.3474056E7|
#|201522369349300227| 178479| 267976| 178479.0|
#|201542399349300619| 1633944| 32850| 1517675.0|
#|201522369349300122| 1401406|1010828|1391213.6666666667|
#|201542399349300724| 1138291|1097553| 1138291.0|
#|201542399349300634| 3402687|1983568| 3691223.5|
#+------------------+--------+-------+------------------+
由于我们将DataFrame连接到自身,因此我们可以使用别名来引用左表("l"
)和右表("r"
)。上面的逻辑说,加入l
到r
,条件是r
的资产是l
资产的+/20%。
有多种表达+ / 20%条件的方法,但我使用spark-sql between
表达式来查找Assets * 0.8
和Assets * 1.2
之间的行。
然后我们在左表的所有列(groupBy
)上聚合,并在右表中的资产上进行平均。
得到的AvgAssets
列是FloatType
列,但你可以通过在IntegerType
之前添加.cast("int")
轻松将其转换为.alias("AvgAssets")
,如果这是你喜欢的。
也可以看看: