如果我没理解错你的问题，基本上你需要-
1.将第一行标记为唯一
1.如果除column_1和column_2外所有列的值都相同，则将所有后续行标记为重复

• • 如果不是，请告诉我**

使用row_number：使用所有列作为分区键进行比较，并为每个分区生成行号，如果一组列值有更多行-它们将属于同一组，并相应地具有row_number。（如果需要，您可以使用orderBy将特定行标记为唯一）：

df.withColumn("asArray", F.array(*[x for x in df.schema.names if x!="column_1" and x!="column_7"]))\
.withColumn("rn", F.row_number().over(Window.partitionBy("asArray").orderBy(F.lit("dummy"))))\
.withColumn("type", F.when(F.col("rn")==1, "Unique").otherwise("Duplicated"))\
.withColumn("tag", F.dense_rank().over(Window.orderBy(F.col("asArray"))))\
.show(truncate=False)

我已经收集了所有列的值，以便在一个数组中进行比较。

• • 编辑-输出与您的数据集相似、重复项超过2个的数据。此外，还更正了标记逻辑**

输入：

输出：

我做了类似这样的事情，这里是输入：

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

data = [
    {"column_1": 34432, "column_2": "apple", "column_3": "banana", "column_4": "mango", "column_5": "pine", "column_6": "lemon", "column_7": "j93jk84"},
    {"column_1": 98389, "column_2": "grape", "column_3": "orange", "column_4": "pine", "column_5": "kiwi", "column_6": "cherry", "column_7": "j93jso3"},
    {"column_1": 94749, "column_2": "apple", "column_3": "banana", "column_4": "mango", "column_5": "pine", "column_6": "lemon", "column_7": "ke03jfr"},
    {"column_1": 48948, "column_2": "grape", "column_3": "banana", "column_4": "mango", "column_5": "pine", "column_6": "lemon", "column_7": "9jef3f4"},
    {"column_1": 90493, "column_2": "pear", "column_3": "apricot", "column_4": "papaya", "column_5": "plum", "column_6": "lemon", "column_7": "93jd30d"},
    {"column_1": 90843, "column_2": "grape", "column_3": "orange", "column_4": "pine", "column_5": "kiwi", "column_6": "cherry", "column_7": "03nd920"}
]

df = spark.createDataFrame(data)

对于第一个df，可以使用dropDuplicates，因为可以传递列的子集

firstDf = df.dropDuplicates(["column_2","column_3","column_4","column_5","column_6"])

对于第二个df，你可以这样做：

windowSpec = Window.partitionBy("column_2","column_3","column_4","column_5","column_6").orderBy("column_1")

secondDf = (
    df.withColumn("row_number", F.row_number().over(windowSpec))
    .withColumn(
        "type",
        F.when(F.col("row_number") == F.lit(1), F.lit("unique")).otherwise(
            F.lit("duplicated")
        ),
    )
    .withColumn("tag", F.first("column_1").over(windowSpec))
    .withColumn("tag", F.dense_rank().over(Window.partitionBy().orderBy("tag")))
).drop("row_number").show()

产出

+--------+--------+--------+--------+--------+--------+--------+----------+---+
|column_1|column_2|column_3|column_4|column_5|column_6|column_7|      type|tag|
+--------+--------+--------+--------+--------+--------+--------+----------+---+
|   34432|   apple|  banana|   mango|    pine|   lemon| j93jk84|    unique|  1|
|   94749|   apple|  banana|   mango|    pine|   lemon| ke03jfr|duplicated|  1|
|   48948|   grape|  banana|   mango|    pine|   lemon| 9jef3f4|    unique|  2|
|   90493|    pear| apricot|  papaya|    plum|   lemon| 93jd30d|    unique|  3|
|   90843|   grape|  orange|    pine|    kiwi|  cherry| 03nd920|    unique|  4|
|   98389|   grape|  orange|    pine|    kiwi|  cherry| j93jso3|duplicated|  4|
+--------+--------+--------+--------+--------+--------+--------+----------+---+