我想收集一个列表，其中包含每个 id1 中具有相同或更低级别的 id2 的所有值。为了实现这一点，我使用窗口函数和 collect_list 函数。但是，我没有得到...

我想收集组内具有相同或较低级别每个 id1 的所有 id2 值的列表。

为了实现这一点，我使用了窗口函数和 collect_list 函数。但是，我在这里没有得到条件部分。如何解决？

df = spark.createDataFrame([
    ("A", 0, "M1", "D1"),
    ("A", 1, "D1", "D2"),
    ("A", 2, "D2", "D3"),
    ("A", 3, "D3", "D4"),
    ("B", 0, "M2", "D5"),
    ("B", 1, "D4", "D6"),
    ("B", 2, "D5", "D7")
], ["group_id", "level", "id1", "id2"])



window = Window.partitionBy('group_id').orderBy('level').rowsBetween(
    Window.unboundedPreceding, Window.unboundedFollowing
)

df_with_list = df.withColumn(
    "list_lower_level",
    F.collect_list("id2").over(window)
)

df_with_list.show()

输出如下：

+--------+-----+---+---+----------------+
|group_id|level|id1|id2|list_lower_level|
+--------+-----+---+---+----------------+
|       A|    0| M1| D1|[D1, D2, D3, D4]|
|       A|    1| D1| D2|[D1, D2, D3, D4]|
|       A|    2| D2| D3|[D1, D2, D3, D4]|
|       A|    3| D3| D4|[D1, D2, D3, D4]|
|       B|    0| M2| D5|    [D5, D6, D7]|
|       B|    1| D4| D6|    [D5, D6, D7]|
|       B|    2| D5| D7|    [D5, D6, D7]|
+--------+-----+---+---+----------------+

然而，我想实现这个目标：

+--------+-----+---+---+----------------+
|group_id|level|id1|id2|list_lower_level|
+--------+-----+---+---+----------------+
|       A|    0| M1| D1|[D1, D2, D3, D4]|
|       A|    1| D1| D2|[D2, D3, D4]|
|       A|    2| D2| D3|[D3, D4]|
|       A|    3| D3| D4|[D4]|
|       B|    0| M2| D5|    [D5, D6, D7]|
|       B|    1| D4| D6|    [D6, D7]|
|       B|    2| D5| D7|    [D7]|
+--------+-----+---+---+----------------+