本文共 1482 字，大约阅读时间需要 4 分钟。

假设您的DataFrame具有类似于以下内容的架构(即Assets和Revenue是数字)：

df.printSchema()

#root

# |-- ID: long (nullable = true)

# |-- Assets: integer (nullable = true)

# |-- Revenue: integer (nullable = true)

您可以在已设置的条件下将DataFrame自身设置为join.加入后,您可以通过对“资产”列的平均值进行分组和汇总.

例如：

from pyspark.sql.functions import avg, expr

df.alias("l")\n .join(

df.alias("r"),

on=expr("r.assets between l.assets*0.8 and l.assets*1.2")

)\n .groupBy("l.ID", "l.Assets", "l.Revenue")\n .agg(avg("r.Assets").alias("AvgAssets"))\n .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”).上面的逻辑说,在r中的资产为l中资产的/ 20％的情况下,将l加入r.

有多种表达/ 20％条件的方法,但我在表达式之间使用spark-sql查找资产* 0.8和资产* 1.2之间的行.

然后,我们对左表的所有列(groupBy)进行汇总,并对右表中的资产进行平均.

生成的AvgAssets列是FloatType列,但是如果您愿意,可以在.alias(“ AvgAssets”)之前添加.cast(“ int”)来轻松地将其转换为IntegerType.

也可以看看：

转载地址：https://blog.csdn.net/weixin_39609822/article/details/110167351 如侵犯您的版权，请留言回复原文章的地址，我们会给您删除此文章，给您带来不便请您谅解！