基于Databricks PySpark按分数和排名筛选30个唯一product_id的问题

核心思路 #

先对每个product_id的所有条目按rank升序（rank值越小排名越高）排序，保留每个ID的前N个高rank条目（比如你例子里的前2个）；接着从这些过滤后的条目中选出30个唯一的product_id（优先选整体rank更高的产品）；最后关联回这些ID的高rank条目，确保最终结果只有30个唯一ID，且每个ID仅保留其高rank的条目。

SQL解决方案 #

假设你的数据存储在product_scores表中，包含product_id、country_gender、score、rank字段：

-- 第一步：给每个product_id的条目按rank排序并标记组内排名
WITH ranked_products AS (
    SELECT 
        product_id,
        country_gender,
        score,
        rank,
        ROW_NUMBER() OVER (PARTITION BY product_id ORDER BY rank ASC) AS group_rank
    FROM product_scores
    WHERE country_gender IN ('US:W', 'US:M', 'CA:W', 'CA:M')
),
-- 第二步：筛选每个product_id的前2个高rank条目
filtered_entries AS (
    SELECT *
    FROM ranked_products
    WHERE group_rank <= 2
),
-- 第三步：选出30个唯一的product_id，按它们的最小rank排序（优先选高排名产品）
top_30_unique_ids AS (
    SELECT DISTINCT product_id
    FROM filtered_entries
    ORDER BY (SELECT MIN(rank) FROM filtered_entries fe WHERE fe.product_id = product_id) ASC
    LIMIT 30
)
-- 第四步：获取最终结果：30个唯一ID对应的所有高rank条目
SELECT fe.*
FROM filtered_entries fe
JOIN top_30_unique_ids ids ON fe.product_id = ids.product_id
ORDER BY fe.product_id, fe.rank ASC;

Python Pandas解决方案 #

如果你的数据是DataFrame格式：

import pandas as pd

# 读取数据（替换成你的数据加载方式）
df = pd.read_csv("your_product_data.csv")

# 1. 筛选目标国家-性别组合的数据
target_groups = ['US:W', 'US:M', 'CA:W', 'CA:M']
filtered_df = df[df['country_gender'].isin(target_groups)]

# 2. 按rank升序排序，保留每个product_id的前2个高rank条目
top_ranked_per_product = filtered_df.sort_values('rank').groupby('product_id').head(2)

# 3. 选出30个唯一的product_id，按产品的最小rank排序（优先选高排名产品）
top_30_ids = top_ranked_per_product.groupby('product_id')['rank'].min().sort_values().head(30).index

# 4. 生成最终结果
final_result = top_ranked_per_product[top_ranked_per_product['product_id'].isin(top_30_ids)]

关键说明 #

• 调整group_rank <= 2或head(2)中的数字，可以控制每个product_id保留的高rank条目数量
• 按产品的最小rank排序选30个ID，能确保你优先拿到整体排名最高的产品，避免低rank产品占用名额

内容的提问来源于stack exchange，提问作者Phani Kumar