如何使用pyspark删除在特定列上具有相同值的重复记录并保留具有最高时间戳的记录

Question

我已经尝试了下面的代码。下面脚本的想法是按 id 和时间戳对记录进行排序，并按处理的时间戳按降序排列，但是当我尝试运行查询时，它并没有按处理的时间戳按降序排列记录。它甚至丢弃了最新记录并保留了不应该的旧记录

df2 = df_concat.orderBy("id", "processed_timestamp", f.col("processed_timestamp").desc()).dropDuplicates(["id"])

我也尝试了下面的方法，但是当我尝试将其转换回数据框时，表架构现在不同了，记录现在位于单个列中并用逗号分隔。它还会丢弃最新记录并保留不应该的旧记录

    def selectRowByTimeStamp(x,y):
        if x.processed_timestamp > y.processed_timestamp:
            return x
        return y

dataMap = df_concat.rdd.map(lambda x: (x.id, x))
newdata = dataMap.reduceByKey(selectRowByTimeStamp) 

我不确定我是否正确理解了上述代码的工作原理。

Answer 1

如果不是因为一个简单的错误，您的代码将按预期工作。

您不应两次使用列名"processed_timestamp"：

df2 = df_concat.orderBy(
    "id", f.col("processed_timestamp").desc()
).dropDuplicates(["id"])

您的代码正在按 processed_timestamp 升序对 DataFrame 进行排序，因为原始列名排在第一位。