python - PySpark。读取 Parquet 时通过转换为字符串来合并模式？

Question

我从 parquet 文件中读取数据，其中有一个 Map 类型字段，如下所示:

>>> df = spark.read.parquet('path/to/partiton')
>>> df.collect()
Row(field={'a': 'SomeString', 'b': '1234'})

>>> df.printSchema()
field: map (containsNull = true)
 |-- key: string
 |-- value: string(valueContainsNull = true)

问题在于，在其他分区中，键 a 为 None，导致键 b 被读取为 long > 类型:

>>> df = spark.read.parquet('path/to/otherPartiton')
>>> df.collect()
Row(field={'a': None, 'b': 1234})

>>> df.printSchema()
field: map (containsNull = true)
 |-- key: string
 |-- value: long(valueContainsNull = true)

同时读取所有分区时，这会产生冲突模式:

>>> df = spark.read.parquet('path/to/')
>>> df.collect()
SparkException: ... java.lang.UnsupportedOperationException: org.apache.parquet.column.values.dictionary.PlainValuesDictionary$PlainLongDictionary

我尝试手动指定架构，如下所示:

>>> struct = StructType([ StructField('field', MapType(StringType(), StringType())) ])
>>> df = spark.read.schema(struct).parquet('path/to/')
>>> df.collect()
fails with same error

有什么办法可以解决这个问题吗？我是否强制重写错误的分区？

Answer 1

您可以在数据帧上强制执行您自己的架构。

架构定义

customschema = StructType([
    StructField("col_01", IntegerType()),
    StructField("col_02", StringType()),
    StructField("col_03", DateType())
])

在您的应用程序代码中，您可以按如下方式进行操作

df = spark.read.csv(filename, header=True, nullValue='NA', schema=customschema)
df.show()