PySpark如何实现支持任意多列的重复值检测自定义函数？

你可以通过修正groupby的参数传递方式来实现需求，核心是正确处理可变参数*args。以下是针对PySpark DataFrame的简便实现：

def find_duplicates(df, *args):
    (df
     .groupby(*args)
     .count()
     .where("count > 1")
     .sort("count", ascending=False)
     .show())

关键说明： #

• *args 会把你传入的所有列名打包成一个元组，比如调用 find_duplicates(df, 'col', 'col2', 'col3') 时，args 就是 ('col', 'col2', 'col3')。
• groupby(*args) 通过解包元组，将每个列名作为单独参数传给分组方法，这样就能正确按指定的多列组合进行分组。
• 后续的count()统计每组的行数（即重复次数），where过滤出重复次数大于1的组，sort按重复次数降序排列，最后用show()展示结果。

如果是针对Pandas DataFrame，实现方式稍有不同：

def find_duplicates(df, *args):
    duplicates_df = (df
                     .groupby(list(args))
                     .size()
                     .reset_index(name='count')
                     .query("count > 1")
                     .sort_values('count', ascending=False))
    print(duplicates_df)

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