Azure Databricks 基于排序过滤的数据去重方案咨询

核心逻辑说明 #

按照code+type作为分组键，每组如果存在2条及以上重复记录，仅保留version不等于PMS的记录；如果组内仅1条记录则直接保留，最终按code、type顺序排序输出即可。 #

实现方式一：PySpark 代码实现 #

# 假设你的原始表名为source_table，先读取为DataFrame
df = spark.table("source_table")

from pyspark.sql import functions as F
from pyspark.sql.window import Window

# 1. 按code+type分组统计每组记录数
window_spec = Window.partitionBy("code", "type")
df_with_cnt = df.withColumn("group_cnt", F.count("*").over(window_spec))

# 2. 按规则过滤记录
df_filtered = df_with_cnt.filter(
    (F.col("group_cnt") == 1) | 
    ((F.col("group_cnt") > 1) & (F.col("version") != "PMS"))
)

# 3. 按要求排序，输出结果
result_df = df_filtered.select("version", "value", "code", "type") \
                       .orderBy(F.col("code").asc(), F.col("type").asc())

# 查看结果
result_df.show()
# 如果要写入表可以执行：result_df.write.saveAsTable("target_table")

实现方式二：Spark SQL 代码实现 #

直接在Databricks的SQL编辑器中执行以下语句即可：

WITH group_count AS (
    SELECT 
        *,
        COUNT(*) OVER (PARTITION BY code, type) AS group_cnt
    FROM source_table -- 替换为你的原始表名
)
SELECT version, value, code, type
FROM group_count
WHERE group_cnt = 1 OR (group_cnt > 1 AND version != 'PMS')
ORDER BY code ASC, type ASC;

输出验证 #

上述两种方式执行后都会得到你要求的结果：

+-------+------+----+---------+
|version| value|code|     type|
+-------+------+----+---------+
|    000|312.00|  01|    HOURS|
|    000|835.00|  01|NON STOCK|
|    000|  0.00|  04|    HOURS|
|    000|835.00|  04|NON STOCK|
+-------+------+----+---------+

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