Spark多列去重性能问题及按最早时间戳保留记录需求

刚好我也碰到过一模一样的需求——需要根据多列判断重复数据，并且只保留时间戳最早的那条记录。亲测下面这个方案完全可行，分享给你：

核心思路 #

利用Spark的orderBy先把时间最早的记录排在前面，再用dropDuplicates指定去重列，它会自动保留排序后的第一条记录（也就是时间最早的那条）。步骤很清晰：

• 把字符串格式的时间列转换成可排序的数值型时间戳
• 按去重列+时间戳升序排序
• 执行去重操作

完整代码示例 #

from pyspark.sql.functions import unix_timestamp
from pyspark.sql import SparkSession

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

# 模拟你的原始数据（替换成你实际的DataFrame即可）
sample_data = [
    ("customer_001", "order_type_A", "2024-05-01 07:20:00"),
    ("customer_001", "order_type_A", "2024-05-01 06:15:00"),
    ("customer_002", "order_type_B", "2024-05-02 15:30:00"),
    ("customer_002", "order_type_B", "2024-05-02 14:45:00")
]
df = spark.createDataFrame(sample_data, ["customer_id", "order_type", "Valid from"])

# 定义时间格式：注意HH是24小时制，hh是12小时制，根据你的实际时间格式调整
time_pattern = "yyyy-MM-dd HH:mm:ss"

# 1. 添加数值型时间戳列，方便排序
df = df.withColumn("timestamp_col", unix_timestamp("Valid from", time_pattern))

# 2. 按去重列（这里是customer_id和order_type）分组后按时间戳升序排序，确保最早的记录在最前面
# 数据量大的话，可以先repartition再排序，减少shuffle开销：df.repartition("customer_id", "order_type").orderBy("timestamp_col")
df_sorted = df.orderBy("customer_id", "order_type", "timestamp_col")

# 3. 指定去重列，执行去重——此时保留的就是排序后的第一条（时间最早的）记录
df_deduped = df_sorted.dropDuplicates(["customer_id", "order_type"])

# 可选：删除临时的时间戳列
df_deduped = df_deduped.drop("timestamp_col")

# 查看去重结果
df_deduped.show()

关键注意点 #

• 时间格式匹配：一定要确保time_pattern和你的Valid from列格式完全一致，比如24小时制用HH，12小时制用hh，否则时间戳转换会失败或者得到错误值
• 性能优化：如果处理的是超大数据集，建议先通过repartition按去重列分区，再进行排序，这样能大幅降低集群的shuffle压力
• dropDuplicates逻辑：这个方法只会保留DataFrame中首次出现的符合去重条件的记录，所以排序的顺序直接决定了哪条记录被保留——升序排序就会留最早的，降序则留最晚的

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