Чтение / запись паркета с типом столбца Struct

Здесь у вас есть как минимум 3 варианта:

Вариант 1:

Вам не нужно использовать какие-либо дополнительные библиотеки, такие как fastparquet, поскольку Spark уже предоставляет эту функцию:

pdf = pd.DataFrame({
    "foo": [1, 2, 3],
    "bar": [{"a": 1, "b": 10}, {"a": 2, "b": 20}, {"a": 3, "b": 30}]
})

df = spark.createDataFrame(pdf)
df.write.mode("overwrite").parquet("/tmp/parquet1")

Если попытаться загрузить данные с помощью df = spark.read.parquet("/tmp/parquet1"), схема будет такой:

StructType([ 
            StructField("foo", LongType(), True),
            StructField("bar",MapType(StringType(), LongType(), True), True)])

Как видите, в этом случае Spark сохранит правильную схему.

Вариант 2:

Если по какой-либо причине по-прежнему необходимо использовать fastparquet, то bar будет рассматриваться как строка, поэтому вы можете загрузить bar как строку, а затем преобразовать ее в JSON, используя from_json. В вашем случае мы будем обрабатывать json как словарь Map (string, int). Это сделано для нашего удобства, поскольку данные кажутся последовательностью ключ / значение, которая может быть прекрасно представлена ​​словарем:

from pyspark.sql.types import StringType, MapType,LongType
from pyspark.sql.functions import from_json

df = spark.read.parquet("/tmp/parquet1")

# schema should be a Map(string, string) 
df.withColumn("bar", from_json("bar", MapType(StringType(), LongType()))).show()

# +---+-----------------+
# |foo|              bar|
# +---+-----------------+
# |  1|[a -> 1, b -> 10]|
# |  2|[a -> 2, b -> 20]|
# |  3|[a -> 3, b -> 30]|
# +---+-----------------+

Вариант 3:

Если ваша схема не меняется и вы знаете, что каждое значение bar всегда будет иметь одинаковую комбинацию полей (a, b), вы также можете преобразовать bar в структуру:

schema = StructType([ 
                    StructField("a", LongType(), True),
                    StructField("b", LongType(), True)
            ])

df = df.withColumn("bar", from_json("bar", schema))

df.printSchema()

# root
#  |-- foo: long (nullable = true)
#  |-- bar: struct (nullable = true)
#  |    |-- a: long (nullable = true)
#  |    |-- b: long (nullable = true)

Пример:

Затем вы можете запустить свой код с помощью:

df.registerTempTable('my_toy_table')

spark.sql("SELECT * FROM my_toy_table WHERE bar.b > 20").show()
# or spark.sql("SELECT * FROM my_toy_table WHERE bar['b'] > 20")

# +---+-----------------+
# |foo|              bar|
# +---+-----------------+
# |  3|[a -> 3, b -> 30]|
# +---+-----------------+