Spark动态分区:读取时出现SchemaColumnConvertNotSupportedException

Question

问题

是否有任何方法可以将具有不同(不兼容(模式的数据存储在不同的分区中

问题

我使用PySpark v2.4.5，镶木地板格式和动态分区，具有以下层次结构：BASE_PATH/COUNTRY=US/TYPE=sms/YEAR=2020/MONTH=04/DAY=10/。不幸的是，它无法改变。

我读到SchemaColumnConvertNotSupportedException。之所以会发生这种情况，是因为模式在不同类型之间(即短信和彩信之间(有所不同。看起来Spark试图在后台读取时合并到架构。

如果更准确地说，我可以读取F.col('TYPE') == 'sms'的数据，因为mms模式可以转换为短信。但当我用F.col('TYPE') == 'mms'过滤时，Spark失败了。

代码

# Works, because Spark doesn't try to merge schemas
spark_session 
.read 
.option('mergeSchema', False) 
.parquet(BASE_PATH + '/COUNTRY_CODE=US/TYPE=mms/YEAR=2020/MONTH=04/DAY=07/HOUR=00') 
.show()
# Doesn't work, because Spark trying to merge schemas for TYPE=sms and TYPE=mms. Mms data can't be converted to merged schema.
# Types are correct, from explain Spark treat date partitions as integers
# Predicate pushdown isn't used for some reason, there is no PushedFilter in explained plan
spark_session 
.read 
.option('mergeSchema', False) 
.parquet(BASE_PATH) 
.filter(F.col('COUNTRY') == 'US') 
.filter(F.col('TYPE') == 'mms') 
.filter(F.col('YEAR') == 2020) 
.filter(F.col('MONTH') == 4) 
.filter(F.col('DAY') == 10) 
.show()

Answer 1

仅针对特定情况，它可能对某些人有用。在不同的分区中可以有不同的数据。要使Spark没有镶木地板的推断模式，请指定模式：

spark_session 
.read 
.schema(some_schema) 
.option('mergeSchema', False) 
.parquet(BASE_PATH) 
.filter(F.col('COUNTRY') == 'US') 
.filter(F.col('TYPE') == 'mms') 
.filter(F.col('YEAR') == 2020) 
.filter(F.col('MONTH') == 4) 
.filter(F.col('DAY') == 10) 
.show()