Spark SQL 读取 Parquet 字段类型不一致异常分析

1. 问题现象

在 Spark SQL 中查询一张 Hive Parquet 表时，任务执行失败，核心异常如下：

org.apache.spark.SparkException: Job aborted due to stage failure:
Task 0 in stage 29.0 failed 4 times

Caused by: java.lang.UnsupportedOperationException:
org.apache.parquet.column.values.dictionary.PlainValuesDictionary$PlainLongDictionary
    at org.apache.parquet.column.Dictionary.decodeToBinary(Dictionary.java:44)
    at org.apache.spark.sql.execution.datasources.parquet.ParquetDictionary.decodeToBinary(ParquetDictionary.java:51)
    at org.apache.spark.sql.execution.vectorized.WritableColumnVector.getUTF8String(WritableColumnVector.java:375)

完整堆栈里比较关键的是这三行：

WritableColumnVector.getUTF8String
ParquetDictionary.decodeToBinary
PlainValuesDictionary$PlainLongDictionary

从表面上看，这个异常很像 Parquet 文件损坏，或者字典编码读取失败。但结合这几个方法名，其实更应该优先怀疑：Spark 读取 Parquet 时使用的字段类型，和 Parquet 文件中该字段真实存储类型不一致。

2. 异常信息怎么理解

先拆开看这几个关键词。

2.1 getUTF8String

getUTF8String 说明 Spark 当前把这个字段当成字符串类型读取。

也就是说，在 Spark 的执行计划里，这一列对应的类型大概率是：

string

或者 Hive Metastore 中该字段被定义成了：

string

2.2 decodeToBinary

字符串在 Parquet 中通常对应的是 BINARY 或 BYTE_ARRAY 这类物理存储。因此 Spark 想从 Parquet 字典中取出二进制内容，再转换成 UTF8 字符串。

所以调用链里出现了：

ParquetDictionary.decodeToBinary

2.3 PlainLongDictionary

但是 Spark 最后拿到的字典类型却是：

PlainLongDictionary

这个名字已经说明底层 Parquet 文件中，这一列的字典值是 long 类型。

于是就出现了一个冲突：

Spark 期望	Parquet 实际
string / binary	long

Spark 想把 long 字典按 binary 字典读取，就会触发：

java.lang.UnsupportedOperationException:
org.apache.parquet.column.values.dictionary.PlainValuesDictionary$PlainLongDictionary

所以这个异常的本质不是 PlainLongDictionary 本身有问题，而是字段类型对不上。

3. 为什么 Hive 表容易出现这个问题

Spark 官方文档中提到，当读取 Hive Metastore 中的 Parquet 表时，Spark SQL 默认会使用自己的 Parquet 支持，而不是 Hive SerDe，这个行为由下面参数控制：

spark.sql.hive.convertMetastoreParquet=true

同时 Spark 读取 Parquet 时默认启用向量化读取：

spark.sql.parquet.enableVectorizedReader=true

这两个默认行为一般是为了提高读取性能，但是在 Hive 表结构和 Parquet 文件真实 schema 不一致时，就容易暴露出问题。

常见场景有几种：

1. Hive 表字段曾经改过类型，例如 bigint 改成了 string。
2. 同一个表目录下存在不同批次写入的 Parquet 文件，字段类型不一致。
3. 上游程序写 Parquet 时使用了 DataFrame schema，但是 Hive 表手工建表时字段类型写错。
4. 分区表中某些历史分区的字段类型和当前表结构不一致。
5. Spark、Hive、Impala 等多个引擎共同写入同一个 Parquet 目录，schema 管理不统一。

Parquet 文件是自描述的，文件 footer 中会保存 schema。Hive Metastore 中也保存了一份表 schema。当这两份 schema 不一致时，Spark 读取 Hive 表就可能在运行时失败。

4. 如何定位是哪一列有问题

4.1 查看 Hive 表结构

先看 Hive Metastore 中记录的表结构：

DESC FORMATTED db_name.table_name;

或者：

SHOW CREATE TABLE db_name.table_name;

重点关注 string 类型字段，尤其是近期改过类型的字段。

4.2 查看 Parquet 文件真实 schema

可以找一个实际数据文件查看 Parquet schema。

如果环境中有 parquet-tools：

parquet-tools schema hdfs://nameservice1/path/to/table/part-00000.parquet

也可以用 Spark 直接读取文件路径，而不是读取 Hive 表：

val df = spark.read.parquet("hdfs://nameservice1/path/to/table")
df.printSchema()

如果是分区表，建议抽几个出问题任务对应的分区单独看：

val df = spark.read.parquet("hdfs://nameservice1/path/to/table/dt=2024-05-22")
df.printSchema()

然后对比：

Hive 表中字段类型
Parquet 文件中字段类型

如果 Hive 中是 string，Parquet 中是 long，就和这次异常吻合。

4.3 缩小问题分区

如果查询的是分区表，可以按分区逐步排查：

SELECT *
FROM db_name.table_name
WHERE dt = '2024-05-22'
LIMIT 10;

如果某些分区失败，某些分区正常，基本可以确定是历史分区或某次写入的数据 schema 不一致。

5. 临时解决方案

5.1 关闭 Parquet 向量化读取

可以先关闭向量化读取验证问题：

set spark.sql.parquet.enableVectorizedReader=false;

或者在提交任务时增加：

spark-submit \
  --conf spark.sql.parquet.enableVectorizedReader=false \
  --class com.xxx.Main \
  your-job.jar

这个参数能绕开一部分向量化读取路径的问题，有时候任务可以先跑过去。

但是需要注意，关闭向量化读取不是根治方案。它只是换了一条读取路径，性能也可能下降。如果底层数据类型和表结构确实不一致，后续仍然可能在其他位置报错，或者出现隐式转换后的错误结果。

5.2 关闭 Hive Metastore Parquet 自动转换

如果这是 Hive 表，可以尝试让 Spark 走 Hive SerDe 读取：

set spark.sql.hive.convertMetastoreParquet=false;

提交任务时可以这样配置：

spark-submit \
  --conf spark.sql.hive.convertMetastoreParquet=false \
  --class com.xxx.Main \
  your-job.jar

这个方式适合临时验证。如果关闭后 Hive 能读，Spark 也能跑，说明问题大概率出在 Spark 内置 Parquet 读取逻辑和 Hive 表 schema、Parquet 文件 schema 的协调上。

但这个方案也会影响性能，并且不建议作为长期方案。

6. 推荐解决方案

6.1 重新生成问题分区

如果确认是某个分区的 Parquet 文件 schema 写错，最稳妥的方式是重新生成这个分区。

例如原来字段实际写成了 long，但 Hive 表定义是 string，可以在重写时显式转换：

import org.apache.spark.sql.functions.col

val fixedDf = sourceDf
  .withColumn("user_id", col("user_id").cast("string"))

fixedDf.write
  .mode("overwrite")
  .parquet("hdfs://nameservice1/path/to/table/dt=2024-05-22")

如果是通过 SQL 写入，也要在写入时显式转换：

INSERT OVERWRITE TABLE db_name.table_name PARTITION(dt='2024-05-22')
SELECT
  cast(user_id as string) as user_id,
  col2,
  col3
FROM tmp_table;

6.2 建一张结构正确的新表

如果历史数据很多，而且多个分区 schema 都不一致，可以考虑重新建表，然后统一写入：

CREATE TABLE db_name.table_name_new (
  user_id string,
  col2 string,
  col3 bigint
)
STORED AS PARQUET;

再用 Spark 或 Hive 统一转换：

INSERT OVERWRITE TABLE db_name.table_name_new
SELECT
  cast(user_id as string) as user_id,
  col2,
  cast(col3 as bigint) as col3
FROM db_name.table_name_old;

6.3 保证写入端 schema 稳定

如果这个表后续还会继续写入，需要从写入端控制 schema：

val result = rawDf.select(
  col("user_id").cast("string").as("user_id"),
  col("event_time").cast("string").as("event_time"),
  col("amount").cast("decimal(18,2)").as("amount")
)

不要依赖 Spark 自动推断类型后直接写入生产表。尤其是 JSON、CSV、Kafka 这类半结构化来源，自动推断很容易因为某个批次的数据内容变化导致字段类型变化。

7. 排查思路总结

遇到下面这种异常时：

WritableColumnVector.getUTF8String
ParquetDictionary.decodeToBinary
PlainValuesDictionary$PlainLongDictionary

可以按这个顺序排查：

步骤	检查内容	目的
1	看异常中的 getUTF8String	判断 Spark 是否把字段当 string 读
2	看异常中的 PlainLongDictionary	判断 Parquet 文件中是否是 long 字典
3	DESC FORMATTED 查看 Hive schema	获取 Metastore 中的字段定义
4	parquet-tools schema 查看文件 schema	获取 Parquet 文件真实字段类型
5	按分区读取	定位是否只有部分分区异常
6	重写问题分区	从根本上修复 schema 不一致

临时参数可以这样用：

set spark.sql.parquet.enableVectorizedReader=false;
set spark.sql.hive.convertMetastoreParquet=false;

但是这两个参数更适合临时验证和应急恢复，不应该替代数据修复。真正要解决问题，还是需要让 Hive 表结构和 Parquet 文件真实 schema 保持一致。

8. 参考资料

• Spark 官方文档：Parquet Files
• Apache Parquet 官方文档：Encodings