PySpark 获取列的最大和最小非零值

Question

我有一个如下的数据框：

+-------+--------------------+--------------------+--------------+---------+----------+
|  label|              app_id|               title|download_count|entity_id|risk_score|
+-------+--------------------+--------------------+--------------+---------+----------+
|ANDROID|com.aaron.test.ze...|          Aaron Test|             0|      124|       100|
|ANDROID|com.boulderdailyc...|Boulder Daily Cam...|           100|      122|       100|
|ANDROID|com.communitybank...|     Budgeting Tools|             0|      123|       100|
|ANDROID|com.communitybank...|   PB Mobile Banking|           600|      123|       100|
|ANDROID|com.mendocinobeac...|Mendocino Beacon ...|            10|      122|       100|
|ANDROID|com.profitstars.t...|Johnson City Mobi...|           500|      123|       100|
|ANDROID|com.spreedinc.pro...|Oneida Dispatch f...|          1000|      122|       100|
+-------+--------------------+--------------------+--------------+---------+----------+

我希望获得按实体 ID 分组的非零最大值和最小值 download_count。我不太清楚如何使用聚合来做到这一点，当然简单的 max 和 min 是行不通的。

apps_by_entity = (
        group_by_entity_id(df)
        .agg(F.min(df.download_count), F.max(df.download_count), F.count("entity_id").alias("app_count"))
        .withColumnRenamed("max(download_count)", "download_max")
        .withColumnRenamed("min(download_count)", "download_min")
    )

因为实体 123 和 124 的最小值将为 0。

+---------+------------+------------+---------+
|entity_id|download_min|download_max|app_count|
+---------+------------+------------+---------+
|      124|           0|           0|        1|
|      123|           0|         600|        3|
|      122|          10|        1000|        3|
+---------+------------+------------+---------+

所需的输出看起来像

+---------+------------+------------+---------+
|entity_id|download_min|download_max|app_count|
+---------+------------+------------+---------+
|      124|           0|           0|        1|
|      123|         500|         600|        3|
|      122|          10|        1000|        3|
+---------+------------+------------+---------+

有没有办法通过聚合来做到这一点？如果不是，那么获得这个非零值的最佳方法是什么？对于max = min = 0，只需返回0 或null 即可。

Answer 1

我不确定您是否可以在进行最小、最大聚合时排除零，而不会丢失计数。
实现输出的一种方法是分别进行 (min, max) 和计数聚合，然后将它们连接回来。

from pyspark.sql.functions import *
from pyspark.sql import functions as F

min_max_df = df.filter(col("download_count")!=0).groupBy('entity_id')\
                .agg(F.min('download_count').alias("download_min"),\
                F.max('download_count').alias("download_max"))\
                .withColumnRenamed("entity_id", "entity_id_1")

count_df =df.groupBy('entity_id').agg(F.count('download_count')\
            .alias("app_count"))

count_df.join(min_max_df, (count_df.entity_id == min_max_df.entity_id_1), \
        "left").drop("entity_id_1").fillna(0, subset=['download_min',\
         'download_max']).show()

+---------+---------+------------+------------+
|entity_id|app_count|download_min|download_max|
+---------+---------+------------+------------+
|      124|        1|           0|           0|
|      123|        3|         500|         600|
|      122|        3|          10|        1000|
+---------+---------+------------+------------+