Сглаживание столбца Maptype в pyspark

Question

У меня есть фрейм данных pyspark со столбцом MapType, и я хочу разбить его на все столбцы по имени ключей

root
 |-- a: map (nullable = true)
 |    |-- key: string
 |    |-- value: long (valueContainsNull = true)

Я хочу сделать sp_test.select('a.*'), но получаю сообщение об ошибке:

AnalysisException: 'Может расширять только типы данных структуры.Атрибут: ArrayBuffer(a); '

Если мы знаем все ключи, это можно сделать, выполнив

sp_test.select(['a.%s'%item for item in ['a','b']]).show()

, но я хотел бы удалить зависимость от ключа

Если у нас есть столбец StrucType, это можно легко сделать, выполнив display(nested_df.select('*', 'nested_array.*'))

root
 |-- _corrupt_record: string (nullable = true)
 |-- field1: long (nullable = true)
 |-- field2: long (nullable = true)
 |-- nested_array: struct (nullable = true)
 |    |-- nested_field1: long (nullable = true)
 |    |-- nested_field2: long (nullable = true)

У меня есть некоторые сомнения:

1. можно ли MapType привести как StructType?
2. можем ли мы запросить субключи напрямую с MapType?

Answer 1

TL; DR: Нет простого способа сделать то, что вы просите, если вы не знаете возможные ключи заранее.

Позвольте мне использовать пример, чтобы попытаться объяснить, почему и каковы ваши варианты.

Сначала создайте следующий фрейм данных:

data = [({'a': 1, 'b': 2},), ({'c':3},), ({'a': 4, 'c': 5},)]
df = spark.createDataFrame(data, ["a"])
df.show()
#+-------------------+
#|                  a|
#+-------------------+
#|Map(a -> 1, b -> 2)|
#|        Map(c -> 3)|
#|Map(a -> 4, c -> 5)|
#+-------------------+

, который имеет следующую схему:

df.printSchema()
#root
# |-- a: map (nullable = true)
# |    |-- key: string
# |    |-- value: long (valueContainsNull = true)

1. Может ли MapType быть приведен как StructType?

Простой ответ - нет (по крайней мере, не эффективно), если вы не знаете ключи заранее.

Разница между MapType и StructType заключается в том, что пары ключ-значение для карт не зависят от строки.Это не относится к столбцу StructType в столбце структуры, все строки имеют одинаковые поля структуры.

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

Итак, если вы знали ключи, вы можете создать тип структуры с помощью:

import pyspark.sql.functions as f

df_new = df.select(
    f.struct(*[f.col("a").getItem(c).alias(c) for c in ["a", "b", "c"]]).alias("a")
)
df_new.show()
#+-------------+
#|            a|
#+-------------+
#|   [1,2,null]|
#|[null,null,3]|
#|   [4,null,5]|
#+-------------+

И новая схема:

df_new.printSchema()
#root
# |-- a: struct (nullable = false)
# |    |-- a: long (nullable = true)
# |    |-- b: long (nullable = true)
# |    |-- c: long (nullable = true)

можем ли мы запрашивать подключи непосредственно из MapType?

Да, (как показано выше) вы можете использовать getItem(), который получит предметпо индексу из списка или по ключу из карты.

---

Если вы не знаете ключей, ваш единственный вариант - explode карта в строки, groupbyи pivot.

df.withColumn("id", f.monotonically_increasing_id())\
    .select("id", f.explode("a"))\
    .groupby("id")\
    .pivot("key")\
    .agg(f.first("value"))\
    .drop("id")\
    .show()
#+----+----+----+
#|   a|   b|   c|
#+----+----+----+
#|null|null|   3|
#|   1|   2|null|
#|   4|null|   5|
#+----+----+----+

В этом случае сначала нужно создать столбец id, чтобы можно было что-то сгруппировать.

Здесь pivot может быть дорогим, в зависимости от размера ваших данных.