如果列最大值小于定义的阈值，则从数据框中删除所有行（对于给定的列值）答案

【问题标题】：Remove all rows (for a given column value) from a dataframe if column max value is less than a defined threshold如果列最大值小于定义的阈值，则从数据框中删除所有行（对于给定的列值）
【发布时间】：2022-11-14 09:03:06
【问题描述】：

如果问题标题有点混乱，请道歉。我是 pyspark 的新手，正在处理以下问题：

假设我有一个数据框，其中包含日期、产品和总订单为三列，我有这个数据框 3 天。所以像

date           product      orders

2022-01-01      whisky        11
2022-01-01      rum           100
2022-01-01      bourbon       5
2022-01-02      whisky        20
2022-01-02      rum           150
2022-01-02      bourbon       7 
2022-01-03      whisky        30
2022-01-03      rum           7
2022-01-03      bourbon       3

我想过滤掉最大订单数小于 10 的任何产品。因此，在上述数据框的情况下，包含波旁威士忌作为产品的所有行都将被过滤掉，因为 max(orders of bourbon) < 10。

输出：

    date       product      orders

2022-01-01      whisky        11
2022-01-01      rum           100
2022-01-02      whisky        20
2022-01-02      rum           150
2022-01-03      whisky        30
2022-01-03      rum           7

最好的方法是什么？我一直在研究 pyspark 中的 Window 功能，但未能正确处理。

我创建了一个这样的windowspec

windowSpec = Window.partitionBy(groupedDf['product']).orderBy(groupedDf['orders'].desc())

但无法过滤掉数据框行。

【问题讨论】：

标签： python pandas pyspark databricks

【解决方案1】：

您可以先找到每个product 的max orders，然后根据该值进行过滤。

df = df.selectExpr('*', 'max(orders) over (partition by product) as max_orders') 
    .filter('max_orders >= 10').drop('max_orders')
df.show(truncate=False)

【讨论】：

嗨，我已经编辑了这个问题，让我的意图更清晰。为混乱道歉。即使朗姆酒在 2022-01-03 的订单少于 10，我仍然需要保留该行作为朗姆酒订单的最大值，即 150 > 10。

【解决方案2】：

这只是窗口函数的情况。

from pyspark.sql.window import Window
import pyspark.sql.functions as F
window = Window.partitionBy("product").orderBy(F.col("orders").desc())
df.withColumn("rn", F.row_number().over(window)).filter("rn < 10")

【讨论】：

嗯不确定它是否有效。它正在从数据集中删除许多不需要的行。原始数据集非常大，按照这个逻辑，我最终删除了应该保留的某些产品行。我想删除最大值不满足阈值的整个产品组，而其他产品组保持不变。
建议的解决方案只做一件事。对于按订单（降序）订购的每个产品，它只会留下前 9 个订单。（需要更改 <= 所以 TOP 10）。
嗨，我已经编辑了这个问题，让我的意图更清晰。为混乱道歉。即使朗姆酒在 2022-01-03 的订单少于 10，我仍然需要保留该行作为朗姆酒订单的最大值，即 150 > 10。

【解决方案3】：

PySparkDataFrame 提供了一个方法toPandas() 将其转换为Python Pandas.DataFrame，然后，首先转换它...

df = pyspark_df.toPandas()

并且您可以通过使用loc 来使用简单的pandas 条件索引：

df = df.loc[df["orders"] > 10]

【讨论】：