python - 在pySpark中处理空数组(可选的二进制元素(UTF8)不是一个组)

Question

我在 spark 中有一个类似 json 的结构，如下所示:

>>> df = spark.read.parquet(good_partition_path)
id: string
some-array: array
    element: struct
        array-field-1: string
        array-field-2: string

取决于分区， some-array可能是所有 id 的空数组的。当这种情况发生时，spark 推断出以下模式:

>>> df = spark.read.parquet(bad_partition_path)
id: string
some-array: array
    element: string

当然，如果我想读取多个分区，这是一个问题，因为 spark 无法合并模式。我试图手动定义架构所以应该没有问题

>>> df = spark.read.schema(good_schema).parquet(bad_partition_path)
id: string
some-array: array
    element: struct
        array-field-1: string
        array-field-2: string

到目前为止一切顺利，但是当我尝试实际收集数据时，出现错误:

>>> df.head(5)
# Long error message
Caused by: java.lang.ClassCastException: optional binary element (UTF8) is not a group

我不明白为什么这会失败。模式应该没有不兼容的原因。如果您想知道，在不指定模式的情况下收集数据是可行的。

>>> df = spark.read.parquet(bad_partition_path)
id: string
some-array: array
    element: string   # infers wrong schema
>>> df.head(5)
[Row(...)] # actually works

编辑

这里有一个在 python 中可重现的例子

from pyspark.sql.types import *


myschema = StructType([
   StructField('id', StringType())
 , StructField( 'some-array'
              , ArrayType(StructType([
                  StructField('array-field-1', StringType())
                , StructField('array-field-2', StringType())
                ])
              ))
  ])

path_writeKO = "path/to/parquet"
jsonKO = '{"id": "OK", "some-array": []}'
dfKO = sc.parallelize([jsonKO])
dfKO = spark.read.json(dfKO)
dfKO.write.parquet(path_writeKO) # write without schema

read_error = spark.read.schema(myschema).parquet(path_writeKO) # read with schema
read_error.collect() # Fails!!

Answer 1

我找到的解决方案是设置选项 dropFieldIfAllNull至 True读取json文件时。这会导致具有空数组的字段消失，从而更容易合并模式。

>>> jsonKO = '{"id": "OK", "some-array": []}'
>>> dfKO = sc.parallelize([jsonKO])
>>> dfKO = spark.read.option('dropFieldIfAllNull', True).json(dfKO)
id:string

现在，不需要的类型推断将不适用，当读取同一文件的多个分区时，选项 mergeSchema将能够无冲突地读取所有文件。