如何在PySpark/SQL中筛选数据库表：保留唯一id，计数大于1时取最小itm_num的记录

需求说明 #

从数据库表中提取记录时，需保留所有唯一的id；对于出现次数大于1的id，需选取其对应itm_num值最小（按升序排序）的记录。

输入数据 #

期望输出数据 #

SQL实现方案 #

用窗口函数ROW_NUMBER()就能轻松搞定，核心思路是给每个id分组内的记录按itm_num升序排名，然后取每组的第一条记录：

WITH ranked_records AS (
    SELECT 
        Source,
        id,
        `group cd`,
        itm_num,
        ROW_NUMBER() OVER (PARTITION BY id ORDER BY itm_num ASC) AS rn
    FROM your_table_name
)
SELECT 
    Source,
    id,
    `group cd`,
    itm_num
FROM ranked_records
WHERE rn = 1;

解释 #

• PARTITION BY id：把相同id的记录归为一组
• ORDER BY itm_num ASC：让每组内itm_num最小的记录排名为1
• 最后筛选rn = 1的记录，既覆盖了重复id取最小itm_num的场景，也自动保留了只出现一次的id（它们的排名本来就是1）

PySpark实现方案 #

和SQL思路完全一致，用PySpark的窗口函数API来实现：

from pyspark.sql import SparkSession
from pyspark.sql.window import Window
from pyspark.sql.functions import row_number

# 初始化SparkSession（如果还没创建的话）
spark = SparkSession.builder.appName("SelectMinItmNumJob").getOrCreate()

# 加载你的数据到DataFrame（示例：从表加载，也可以从CSV/JSON等数据源读取）
# df = spark.read.table("your_table_name")

# 定义窗口规则：按id分区，按itm_num升序排序
window_spec = Window.partitionBy("id").orderBy("itm_num")

# 生成排名列并筛选目标记录
result_df = df.withColumn("rn", row_number().over(window_spec)) \
              .filter("rn = 1") \
              .drop("rn")

# 查看结果或者保存
result_df.show()
# result_df.write.mode("overwrite").saveAsTable("your_result_table")

解释 #

• 先通过Window类定义分组和排序规则，然后用row_number()生成每条记录在组内的排名
• 过滤掉排名不为1的行，再删除辅助的rn列，就得到了符合需求的结果

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