Как застегнуть две колонки в pyspark?

Question

Я использую: Python 3.6 и PySpark 2.3.0.В следующем примере у меня есть только объекты буксировки в item, но также я могу получить больше информации, например first_name, last_name, city.

У меня есть фрейм данных со следующей схемой:

|-- email: string (nullable = true)
| -- item: struct(nullable=true)
| | -- item: array(nullable=true)
| | | -- element: struct(containsNull=true)
| | | | -- data: string(nullable=true)
| | | | -- fieldid: string(nullable=true)
| | | | -- fieldname: string(nullable=true)
| | | | -- fieldtype: string(nullable=true)

Это мой вывод:

+-----+-----------------------------------------------------------------------------------------+
|email|item                                                                                     |
+-----+-----------------------------------------------------------------------------------------+
|x    |[[[Gmail, 32, Email Client, dropdown], [Device uses Proxy Server, 33, Device, dropdown]]]|
|y    |[[[IE, 32, Email Client, dropdown], [Personal computer, 33, Device, dropdown]]]          |
+-----+-----------------------------------------------------------------------------------------+

Я хочу преобразовать этот фрейм данных в:

+-----+-------------------------------------+
|email|Email Client|Device                  |
+-----+-------------------------------------+
|x    |Gmail       |Device uses Proxy Server|
|y    |IE          |Personal computer       |
+-----+-------------------------------------+

Я делаю некоторые преобразования:

df = df.withColumn('item', df.item.item)
df = df.withColumn('column_names', df.item.fieldname)
df = df.withColumn('column_values', df.item.data)

А теперь мой вывод:

+-----+----------------------+---------------------------------+
|email|column_names          |column_values                    |
+-----+----------------------+---------------------------------+
|x    |[Email Client, Device]|[Gmail, Device uses Proxy Server]|
|y    |[Email Client, Device]|[IE, Personal computer]          |
+-----+----------------------+---------------------------------+

Здесь я хочу метод, как сжать эти столбцы.

Answer 1

Вы спрашивали, как zip массивы, но на самом деле вы можете получить желаемый результат без промежуточных шагов создания столбцов column_names и column_values.

Используйте getItem() функция для захвата желаемых значений по индексу:

import pyspark.sql.functions as f
df = df.select(
    'email',
    f.col('item.data').getItem(0).alias('Email Client'),
    f.col('item.data').getItem(1).alias('Device')
)
df.show(truncate=False)
#+-----+------------+------------------------+
#|email|Email Client|Device                  |
#+-----+------------+------------------------+
#|x    |Gmail       |Device uses Proxy Server|
#|y    |IE          |Personal computer       |
#+-----+------------+------------------------+

Предполагается, что поле Email Client всегда имеет индекс 0, а Device - индекс 1.

---

Если вы не можете предположить, что поля в каждом ряду всегда находятся в одинаковом порядке, другой вариант - создать карту из значений в column_names и column_values, используя pyspark.sql.functions.create_map().

Эта функция принимает:

список имен столбцов (строка) или список выражений столбцов, которые [сгруппированы] как пары ключ-значение, например (key1, значение1, ключ2, значение2, ...).

Мы перебираем элементы в column_names и column_values, чтобы создать список пар, а затем используем list(chain.from_iterable(...)) для выравниваниясписок.

После составления списка вы можете выбрать поле по имени.

from itertools import chain

# first create a map type column called 'map'
df.select(
    'email',
    f.create_map(
        list(
            chain.from_iterable(
                [[f.col('column_names').getItem(i), f.col('column_values').getItem(i)] 
                 for i in range(2)]
            )
        )
    ).alias('map')
)
df.show(truncte=False)
#+-----+--------------------------------------------------------------+
#|email|map                                                           |
#+-----+--------------------------------------------------------------+
#|x    |Map(Email Client -> Gmail, Device -> Device uses Proxy Server)|
#|y    |Map(Email Client -> IE, Device -> Personal computer)          |
#+-----+--------------------------------------------------------------+

# now select the fields by key
df = df.select(
    'email',
    f.col('map').getField("Email Client").alias("Email Client"),
    f.col('map').getField("Device").alias("Device")
)

Предполагается, чтоВ каждом массиве всегда должно быть не менее 2 элементов.

---

Если вы хотите сжать списки произвольной длины, вам придется использовать udf.

# define the udf
zip_lists = f.udf(lambda x, y: [list(z) for z in zip(x, y)], ArrayType(StringType()))

# use the udf to zip the lists
df.select(
    'email',
    zip_lists(f.col('column_names'), f.col('column_values')).alias('zipped')
).show(truncate=False)
#+-----+-----------------------------------------------------------+
#|email|zipped                                                     |
#+-----+-----------------------------------------------------------+
#|x    |[[Email Client, Gmail], [Device, Device uses Proxy Server]]|
#|y    |[[Email Client, IE], [Device, Personal computer]]          |
#+-----+-----------------------------------------------------------+

илиВы можете использовать udf для создания карты:

make_map = f.udf(lambda x, y: dict(zip(x, y)), MapType(StringType(), StringType()))
df.select(
    'email',
    make_map(f.col('column_names'), f.col('column_values')).alias('map')
).show(truncate=False)
#+-----+--------------------------------------------------------------+
#|email|map                                                           |
#+-----+--------------------------------------------------------------+
#|x    |Map(Device -> Device uses Proxy Server, Email Client -> Gmail)|
#|y    |Map(Device -> Personal computer, Email Client -> IE)          |
#+-----+--------------------------------------------------------------+