PySpark - Как превратить повторяющиеся строки в новые столбцы

Question

Допустим, у нас есть таблица clients вида:

| id | Name |
|-----------|
| 1  | John |
| 2  | Bob  |
| 3  | Ella |
| 4  | Jim  |

и таблица vehicles для каждого клиента:

| id |client_id| vehicle |
|------------------------|
| 1  |    1    |  car1   |
| 2  |    2    |  car2   |
| 3  |    2    |  car3   |
| 4  |    2    |  car4   |

Теперь мы можем видеть что у Боба 3 машины. Я хотел бы добавить эти транспортные средства в таблицу clients таким образом, чтобы они получали новый столбец для каждого транспортного средства с соответствующим client_id.

Это должно выглядеть примерно так:

| id | Name | vehicle1 | vehicle2 | vehicle3 |
|--------------------------------------------|
| 1  | John |   car1   |   null   |   null   |
| 2  | Bob  |   car2   |   car3   |   car4   |
| 3  | Ella |   null   |   null   |   null   |
| 4  | Jim  |   null   |   null   |   null   |

Можно ли этого достичь?

Answer 1

Другой подход:

Просто обратите внимание, что может быть немного медленным Поскольку мы оцениваем один и тот же набор данных дважды (сначала для определения максимальной длины, а затем для получения окончательного набора данных с использованием максимальной длины)

import pyspark.sql.functions as f
df = df1.join(df2, [df1.id == df2.client_id], 'left_outer').groupBy(df1['id'],'Name').agg(f.collect_list('vehicle').alias('vehicle'))
df.show()
+---+----+------------------+
| id|Name|           vehicle|
+---+----+------------------+
|  1|John|            [car1]|
|  3|Ella|                []|
|  2|Bob |[car3, car4, car2]|
|  4|Jim |                []|
+---+----+------------------+

Найти максимальную длину для всех транспортных средств и получить окончательный набор данных

max_len = df.select(f.max(f.size('vehicle')).alias('max')).first()['max']


df.select('id', 'Name', *[df.vehicle[x] for x in range(max_len)]).show()
+---+----+----------+----------+----------+
| id|Name|vehicle[0]|vehicle[1]|vehicle[2]|
+---+----+----------+----------+----------+
|  1|John|      car1|      null|      null|
|  3|Ella|      null|      null|      null|
|  2|Bob |      car2|      car3|      car4|
|  4|Jim |      null|      null|      null|
+---+----+----------+----------+----------+

Answer 2

ПРЕДУПРЕЖДЕНИЕ: Для этого решения требуются данные, к которым применяется udf_numbering, либо для одного исполнителя, либо, по крайней мере, каждый идентификатор полностью содержится в одном исполнителе. При запуске на нескольких исполнителях, где данные случайным образом распределены между ними, каждый получит свою собственную копию id_dict без учета изменений на других исполнителях.

Понял. Ужасно медленно, но делает работу *

clients_df = clients_df.join(vehicles_df, vehicles_df('client_id') == clients_df('id'), 'left')