Установить союз над окном в PySpark

Question

Я пытаюсь получить список всех уникальных значений, которые появляются для определенной группы идентификаторов, и все это за окном. Что касается данных ниже, для каждого address я хочу найти все id, которые имеют общие value и собрать набор всех value, которые появляются для группы id.

Например, у меня есть window.partitionBy('address'). Для address: 1 я вижу, что id: A, B, C имеет общее value: x. Я понимаю, что они связаны и хотят создать value_set со всеми value, которые соответствуют id: A, B, C, то есть x,y,z.

id: D не имеет общего value с любым другим id, поэтому value_set включает только значения id: D.

My data
+-------+---+-----+
|address| id|value|
+-------+---+-----+
|      1|  A|    x|
|      1|  A|    у|
|      1|  B|    x|
|      1|  C|    x|
|      1|  C|    z|
|      1|  D|    v|
|      2|  E|    m|
|      2|  E|    n|
|      2|  F|    m|
|      2|  F|    p|
+-------+---+-----+

What I want
+-------+---+-----+---------+
|address| id|value|value_set|
+-------+---+-----+---------+
|      1|  A|    x|    x,y,z|
|      1|  A|    у|    x,y,z|
|      1|  B|    x|    x,y,z|
|      1|  C|    x|    x,y,z|
|      1|  C|    z|    x,y,z|
|      1|  D|    x|    v    |
|      2|  E|    m|    m,n,p|
|      2|  E|    n|    m,n,p|
|      2|  F|    m|    m,n,p|
|      2|  F|    p|    m,n,p|
+-------+---+-----+---------+

Answer 1

как то так?

dfn = df.withColumn('set_collect',collect_set(df.value).over(Window.partitionBy('address')))

Вывод:

+-------+---+-----+------------+
|address| id|value| set_collect|
+-------+---+-----+------------+
|      1|  A|    x|[у, v, z, x]|
|      1|  A|    у|[у, v, z, x]|
|      1|  B|    x|[у, v, z, x]|
|      1|  C|    x|[у, v, z, x]|
|      1|  C|    z|[у, v, z, x]|
|      1|  D|    v|[у, v, z, x]|
|      2|  E|    m|   [n, m, p]|
|      2|  E|    n|   [n, m, p]|
|      2|  F|    m|   [n, m, p]|
|      2|  F|    p|   [n, m, p]|
+-------+---+-----+------------+