PySpark中dropDuplicates报错：参数应为列表/元组而非DataFrame

问题根源 #

你调用df.dropDuplicates(df)时犯了参数类型错误：PySpark的dropDuplicates()方法要求subset参数是列名的列表或元组（用来指定按哪些列去重），但你直接传入了整个DataFrame对象，这就触发了类型不匹配的报错。

正确用法示例 #

dropDuplicates()有两种常用调用方式：

• 按所有列去重：不需要传参数，直接调用

  df.dropDuplicates()
  

• 按指定列去重：传入列名组成的列表/元组

  # 比如按"user_id"和"order_date"两列去重
  df.dropDuplicates(["user_id", "order_date"])
  

修正你的去重函数 #

假设你的原函数是这样的：

def drop_duplicates(df):
    # 错误写法：把df传给了subset参数
    return df.dropDuplicates(df)

修正后的函数可以支持默认全列去重，也能指定列：

def drop_duplicates(df, subset=None):
    if subset is not None:
        # 确保subset是列表/元组类型
        if not isinstance(subset, (list, tuple)):
            raise TypeError("subset必须是列表或元组类型")
        return df.dropDuplicates(subset)
    # 不传subset时默认按所有列去重
    return df.dropDuplicates()

测试用例示例（UnitTest） #

import unittest
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, IntegerType

class TestDropDuplicates(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.spark = SparkSession.builder.appName("TestDropDuplicates").getOrCreate()

    @classmethod
    def tearDownClass(cls):
        cls.spark.stop()

    def test_full_columns_duplicate_drop(self):
        # 构造含重复数据的测试DataFrame
        schema = StructType([
            StructField("id", IntegerType(), True),
            StructField("name", StringType(), True)
        ])
        data = [(1, "Alice"), (2, "Bob"), (1, "Alice"), (3, "Charlie")]
        df = self.spark.createDataFrame(data, schema)
        
        # 调用去重函数
        deduplicated_df = drop_duplicates(df)
        
        # 验证结果：原4条数据去重后剩3条
        self.assertEqual(deduplicated_df.count(), 3)

    def test_specified_columns_duplicate_drop(self):
        data = [(1, "Alice"), (1, "Bob"), (2, "Alice"), (1, "Bob")]
        df = self.spark.createDataFrame(data, ["id", "name"])
        
        # 按id列去重
        deduplicated_df = drop_duplicates(df, subset=["id"])
        
        # 验证结果：去重后剩2条（id=1和id=2）
        self.assertEqual(deduplicated_df.count(), 2)

if __name__ == "__main__":
    unittest.main()

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