PySpark合并DataFrame行：保留包含子集的最大数组行

Got it, let's work through this problem step by step to get you from the sorted/distinct DataFrame to the final result where only the "maximal" arrays (those that aren't subsets of any other array) are kept.

Step 1: Recap Your Current Progress #

First, you already nailed the deduplication of permutations by sorting arrays and taking distinct values—great start! Here's that code again for context:

from pyspark.sql import functions as f

df = spark.createDataFrame([
    (["1", "2","3","4"], ),
    (["1","2","3"], ),
    (["2","1","3"], ),
    (["2","3","4","1"], ),
    (["6","7"], )
], ['cycle', ])

# Get sorted, distinct arrays to eliminate permutations
sorted_df = df.select(f.sort_array("cycle").alias("cycle")).distinct()
sorted_df.show()

Which outputs:

+------------+
|       cycle|
+------------+
|[1, 2, 3, 4]|
|     [6, 7]|
| [1, 2, 3]|
+------------+

Step 2: Remove Subset Arrays with Distributed Operations #

The goal now is to filter out arrays that are subsets of longer arrays—without resorting to slow, non-distributed row-by-row loops. The core logic is: Keep an array only if there is no other longer array that contains all its elements.

Here's how to implement this efficiently with PySpark:

1. Self-Join to Identify Subsets: Join the sorted DataFrame with itself to find pairs where one array is a subset of another and shorter.
2. Left Anti Join to Filter Out Subsets: Use a left anti join to retain only arrays that don't have any longer "parent" array containing them.

# Self-join to find arrays that are subsets of longer arrays
subset_pairs = sorted_df.alias("a").join(
    sorted_df.alias("b"),
    # Check if array a is a subset of array b (array_except returns empty)
    (f.size(f.array_except("a.cycle", "b.cycle")) == 0) &
    # Ensure array a is shorter than array b (avoids matching identical arrays)
    (f.size("a.cycle") < f.size("b.cycle"))
)

# Left anti join to keep only arrays that are NOT subsets of any longer array
result_df = sorted_df.join(
    subset_pairs.select("a.cycle").alias("to_remove"),
    on="cycle",
    how="left_anti"
)

result_df.show()

Final Output #

Running this code will give you exactly the result you're after:

+------------+
|       cycle|
+------------+
|[1, 2, 3, 4]|
|     [6, 7]|
+------------+

Why This Works #

• array_except(a.cycle, b.cycle) returns an empty array if every element in a.cycle exists in b.cycle—this confirms a is a subset of b.
• The left anti join keeps all rows in sorted_df that don't appear in the to_remove list (which contains all subset arrays).
• This approach is fully distributed, so it scales efficiently even for large datasets—no slow Python loops involved.

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