Spark SQL 读取 Parquet 失败排查：Schema 不一致

spark

发布日期: 2022-10-23

文章字数: 4.8k

阅读时长: 18 分

阅读次数:

一、问题现象

在执行 Spark SQL 查询 Parquet 表时，任务直接失败，并出现如下异常：


org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 29.0 failed 4 times, most recent failure: Lost task 0.3 in stage 29.0 (TID 1097, rzxgzsj02, executor 2): 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)
	at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)
	at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
	....
Driver stacktrace:
	at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1651)
	at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1639)
	at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1638)
	at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
	at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
	at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1638)
	at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:831)
	...
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)
	....
org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 29.0 failed 4 times, most recent failure: Lost task 0.3 in stage 29.0 (TID 1097, rzxgzsj02, executor 2): 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)
	at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)
	....

Driver stacktrace:
	at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1651)
	at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1639)
	at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply
    ....
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)
	at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)
	at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)

单看这个报错信息，其实很难直接判断问题本质。但结合完整的堆栈信息，可以发现几个非常关键的调用路径：

WritableColumnVector.getUTF8String
ParquetDictionary.decodeToBinary
Dictionary.decodeToBinary

这些方法串联起来，已经明确反映了当前的执行逻辑：

Spark 正在尝试将某个字段按字符串类型读取
在读取过程中进入了 getUTF8String() 路径
底层数据却来自 PlainLongDictionary（long 类型的字典编码）
在进行字符串解码时发生类型不匹配，最终抛出异常

因此，这个问题并不是执行过程中的偶发失败，也不是资源或环境问题，而是一个非常典型的读取逻辑与底层数据类型不一致的问题。

从本质上看，Spark 将该字段当作 string 类型处理，但 Parquet 文件中该字段实际存储为 long / bigint 类型，在解码阶段发生了冲突，最终导致任务失败。

在生产环境中，这类问题通常具有比较明显的特征：

表历史上可以正常查询，但在某个时间点之后开始报错
查询整表失败，但部分分区单独查询可能正常
报错看起来像 Parquet 编码或兼容性问题，但根因通常是 schema 发生变化
调整读取参数（例如关闭 vectorized reader）后表现可能发生变化，但问题本身并没有消失

这类问题的核心不在于“执行失败”，而在于数据 schema、存储结构和读取逻辑之间出现了不一致。

二、从原理理解这个错误

很多 Spark 错误如果只看异常信息，会让人感觉很随机。但实际上，只要把底层读取流程想清楚，这个错误是非常有规律的。

1. Parquet 的存储机制

Parquet 是一种列式存储格式。所谓列式存储，意思是：

一张表的数据不是按“整行”连续存储
而是按“列”分别存储
每一列都有自己的物理类型、编码方式和统计信息

例如，一张表里有如下字段：

id bigint
name string
age int

那么在 Parquet 文件中：

id 这一列会按它自己的物理类型单独存储
name 这一列也会按它自己的物理类型单独存储
age 同样如此

这样做的好处是：

压缩率高
查询某几列时不需要扫描整行
对分析型场景非常友好

同时，Parquet 为了进一步提升压缩率，还会使用多种编码方式，其中一种非常常见的就是 Dictionary Encoding（字典编码）。

你可以把字典编码理解成：

先把某列中重复出现的值收集起来形成字典
实际存储时不直接写原始值，而是写字典索引

如果某一列是 long 类型，那么字典中存储的也是 long 值，对应的实现就可能是：

PlainLongDictionary

也就是说，看到 PlainLongDictionary 并不说明 Parquet 文件坏了，恰恰说明：

底层这列本来就是按 long 类型并且用了字典编码来存储。

2. Spark 是如何读取 Parquet 的

Spark 读取 Parquet 时，并不是“看到什么就读什么”，它实际上同时依赖两套信息：

读取端 schema
- 可能来自 Hive Metastore 表定义
- 也可能来自 DataFrame 的显式 schema
- 或者来自 Spark 推断出的 schema
Parquet 文件实际 schema
- 来自底层文件 footer 中保存的真实字段类型信息

正常情况下，这两套 schema 应该是一致的。

比如：

Spark 认为 id 是 bigint
Parquet 文件里 id 也确实是 INT64

那么 Spark 就能顺利读取。

但如果出现下面这种情况：

Spark 认为 id 是 string
Parquet 文件里 id 实际是 INT64

那读取时就会出问题。

3. 为什么会报 `getUTF8String -> decodeToBinary -> PlainLongDictionary`

现在回到你的报错链路：

getUTF8String → decodeToBinary → PlainLongDictionary

这条链路非常关键。

第一步：Spark 调用了 `getUTF8String`

这说明 Spark 当前把这个字段当成了字符串列来读取。

因为如果 Spark 认为它是 long，通常就会走数值类型的读取路径，而不是 UTF8String。

第二步：Spark 进一步调用 `decodeToBinary`

这说明 Spark 期望底层字典能够“解码成二进制/字符串表示”。

对于字符串类型，底层通常会以类似 BYTE_ARRAY 的形式存在，这样解码成字符串是合理的。

第三步：底层真实对象却是 `PlainLongDictionary`

这一步就彻底暴露问题了。

底层数据字典根本不是字符串字典，而是 long 字典。也就是说，Spark 预期的是“把这个字段当字符串读”，但真实数据是“一个 long 编码列”。

于是最终结果就是：

预期：字符串读取路径
实际：long 类型字典
结果：类型不兼容，抛异常

所以这个错误并不是 Parquet “不能读 long”，而是：

Spark 用错了读法。

而 Spark 为什么会用错读法？答案就是：

Schema 不一致

这也是这类问题最核心的根因。

三、常见触发场景

知道原理之后，再看线上问题就会非常清楚。这个错误几乎不会无缘无故出现，通常都对应某种数据演化或元数据失真。

场景 1：Hive 表结构与 Parquet 文件不一致

这是最常见的场景之一。

例如：

Hive 表字段定义成了 string
但底层 Parquet 文件实际写入的是 bigint

这类问题很容易出现在：

建表时手工写错字段类型
开发同学参考旧表结构复制建表语句时没改全
同一路径底层文件被其他任务按不同 schema 写入

表面上看，Hive 表能正常创建、字段名也都对，但只要字段类型和底层物理类型不匹配，Spark 在扫描时就可能炸掉。

这类问题最麻烦的地方在于：

表 schema 看起来“合法”
Parquet 文件本身也“没有坏”
但两边拼在一起就出错

场景 2：同一张表不同分区类型不一致

这也是生产环境非常典型的问题。

例如分区表按天写入：

dt=2023-01-01 分区中，某列是 bigint
dt=2023-02-01 分区中，同一列被写成了 string

如果你只查一个正常分区，可能不会报错；但一旦：

查整张表
或者查跨多个分区
或者 Spark 触发 schema merge / 统一读取逻辑

就会把这些不一致的分区放到一起解释，最终报错。

这种问题最容易发生在：

分区长期由不同版本程序写入
历史代码没有严格锁定 schema
中间层数据清洗逻辑发生了变化

所以当你发现“某张表有时候能查、有时候不能查”时，一定要高度怀疑是不是分区之间 schema 演化失控了。

场景 3：历史数据与新版本代码的字段类型变更

这个场景本质上是“数据演进管理失控”。

比如最初写入逻辑是：

user_id: long

后来由于业务改造，开发把它改成：

user_id: string

如果只是代码改了，而历史数据没有回刷、目标路径没有切换，那么同一路径下就会混合存在两种类型的数据文件。

对写入侧来说，这种改动可能看起来只是“小改动”；但对读取侧来说，这是一个非常严重的 schema 兼容性问题。

尤其在数据湖/ODS 层长期累积之后，这种问题会非常隐蔽：

老分区是旧 schema
新分区是新 schema
表定义可能还停留在旧版本

最后，读取侧就会在某一天突然爆炸。

场景 4：直接写 Parquet 路径，再通过外部表读取

例如有些任务会这样做：

// 写入 parquet 文件
 df.write.parquet("/path/data")

然后再单独建一个外部表：

CREATE EXTERNAL TABLE ...
LOCATION '/path/data';

这种做法本身没问题，但前提是：

外部表 schema 必须和文件真实 schema 完全一致。

如果外部表字段类型定义错了，就会出现“元数据层认为是 string，文件层实际是 bigint”这种情况。

这类问题在线上很常见，因为建表和写数的人不一定是同一个团队，最终就会出现：

数据路径是真实的
表也存在
但读出来就报错

场景 5：多数据源汇总写入同一路径

还有一种场景是，多个来源的数据在落地前没有做严格的 schema 对齐。

比如：

来源 A 的 order_id 是 bigint
来源 B 的 order_id 是 string
最终都 append 到同一张 Parquet 表路径

这种情况在数据接入平台、ODS 汇总层、埋点日志接入层特别容易发生。

如果写入前没有强制 cast 和 schema 校验，后面就非常容易在 Spark 查询时触发这类报错。

四、如何定位问题

知道了原理和触发场景之后，接下来最关键的是：

如何高效定位到底是哪一列、哪一个分区、哪一批文件有问题。

这一部分决定了排查效率。

1. 先看表结构

第一步先确认读取端认为这张表长什么样。

DESC FORMATTED table_name;

重点关注：

报错涉及字段的类型
表是否为外部表
表对应的 LOCATION
分区字段定义

如果这里已经看到某个字段被定义为 string，而你业务上知道它本来应该是数值型，那就要高度警惕。

不过注意：

DESC FORMATTED 只能告诉你“表怎么定义”，不能证明“文件一定就是这样”。

所以这一步只是第一步。

2. 查看 Spark 侧 DataFrame schema

如果你是通过 Spark 表读取，可以继续：

spark.table("table_name").printSchema()

这一步的意义是确认：

Spark 当前真正拿到的 schema 是什么
有没有经过 catalog / 推断之后发生变化

有时候 Hive 表定义和 Spark 实际解析出来的 schema 并不完全一样，尤其是在某些兼容配置下，printSchema 可以帮助你确认 Spark 最终使用的是哪套结构。

3. 直接读底层 Parquet 文件

这是最关键的一步之一。

spark.read.parquet("/path/to/data").printSchema()

这一步是为了看：

底层 Parquet 文件自己声明的真实 schema 到底是什么。

到这里，你就可以开始做最核心的对比了：

表 schema 是什么
Spark 读取表时看到的 schema 是什么
底层 parquet 文件真实 schema 是什么

如果三者不一致，问题基本就锁定了。

4. 按分区缩小范围

如果这是一张分区表，直接查整表通常信息量太大。更推荐的方法是逐个分区试。

SELECT * FROM table_name WHERE dt = '2023-01-01' LIMIT 10;

通过这种方式，你可以逐步缩小问题范围：

哪些分区能查
哪些分区一查就报错
错误是从哪天开始出现的

一旦定位到异常分区，再直接读对应分区路径：

spark.read.parquet("/path/to/table/dt=2023-01-01").printSchema()

这样往往就能非常快地发现：

某个分区字段类型变了
或者某批文件结构发生了漂移

这一步在生产排查中非常有效，因为大部分 schema 漂移问题都不是“全表同时坏掉”，而是从某个时间点、某个分区开始出现。

5. 关闭向量化读取辅助定位

set spark.sql.parquet.enableVectorizedReader=false;

这一步不是为了解决根因，而是为了验证问题是否和向量化读取路径有关。

如果关闭后：

查询能跑通
或者报错形式发生变化

那么通常说明：

底层数据确实存在类型问题
vectorized reader 在当前场景下更容易触发这个异常

它的意义主要有两个：

帮助你确认问题不是“资源/网络/任务调度”层面的
作为临时手段辅助读取并进一步分析数据

但要强调的是：

关闭 vectorized reader 不是根治，只是辅助定位或短期规避。

6. 必要时直接读单个文件

如果分区里也混了多种 schema，甚至同一分区下只有个别文件坏掉，那么还需要继续缩小到单文件级别。

这一步通常适用于：

历史回灌
多任务并发写入
某次异常任务只污染了一小批文件

通过逐步读取更小范围的数据路径，可以定位到具体异常文件。

五、解决方案

定位清楚后，修复方案就会比较明确。一般来说，解决这类问题的核心思路只有一个：

让读取端 schema 和底层文件 schema 保持一致。

但具体做法取决于问题在哪一层。

方案一：修正表 schema（推荐）

如果确认底层 Parquet 文件的类型是正确的，而表定义错了，那么最直接的修复方式就是修改表 schema。

例如：

ALTER TABLE table_name CHANGE COLUMN col_name col_name BIGINT;

这个方案适用于：

文件一直是对的
只是 Hive/外部表定义错了
全部历史分区都保持相同物理类型

它的优点是：

成本低
不需要重写数据
修复速度快

但要注意一个前提：

必须确认所有底层文件都确实是这个类型。

否则你把表 schema 改过来以后，可能只是把当前错误换成另一个错误。

方案二：重写异常分区

如果问题只存在于部分分区，那么最稳妥的方式往往是重写这些异常分区。

例如：

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

val df = spark.read.parquet("/bad_partition")

// 统一字段类型
val newDf = df.withColumn("col_name", col("col_name").cast("string"))

newDf.write.mode("overwrite").parquet("/bad_partition")

这个方案的核心思想是：

先把坏分区读出来
按目标 schema 做显式类型转换
再覆盖写回去

它适用于：

少量分区异常
分区间 schema 不一致
历史数据中混入了错误类型

相较于直接改表 schema，这个方案更重，但也更稳。

方案三：统一写入逻辑（根治）

如果问题反复出现，那就说明不是单次事故，而是数据生产链路本身有问题。

这时不能只修消费端，而应该在写入侧根治。

需要做到：

字段类型不随版本随意变化
写入前统一 cast
多来源数据汇总前统一 schema
在写入任务中增加 schema 校验

例如在写入前就强制：

val normalizedDf = rawDf
  .withColumn("col_name", col("col_name").cast("bigint"))

这类治理措施看起来“麻烦”，但它们是从根源上避免此类问题再次发生的关键。

方案四：临时规避（不推荐长期使用）

set spark.sql.parquet.enableVectorizedReader=false;

这个方案适合：

紧急恢复查询
临时做数据排查
在根修复完成前先让任务跑通

但它的局限也很明确：

只是 workaround
不能消除 schema 不一致
还可能影响查询性能

所以它只能作为临时手段，不适合作为长期方案。

方案五：重建表或切换新路径

如果历史包袱太重，比如：

同一路径混了太多版本的数据
表 schema 多次变更
分区大量污染

那么最干净的方式有时不是修修补补，而是：

按统一 schema 写入新路径
重新建表
让新表只承接清洗后的规范数据

这在生产中也是非常常见的决策。

因为当历史数据已经非常脏时，继续在原路径上修可能成本更高。

六、总结

这个问题如果只看报错名称，会显得很奇怪；但从原理上讲，它其实非常简单：

Spark 读取 schema 与 Parquet 实际 schema 不一致

具体表现就是：

Spark 把字段当成 string 去读
底层实际却是 long/bigint
最终在字典解码阶段失败

所以，看到下面这些关键词时：

getUTF8String
decodeToBinary
PlainLongDictionary

第一反应就应该是：

先查 schema，不要先怀疑机器。

最佳实践

为了尽量避免这类问题，生产环境里建议做到：

表 schema 和底层文件 schema 保持一致
避免历史版本随意修改字段类型
写入前统一 schema
多数据源汇总时先做类型对齐
对分区表定期做 schema 巡检

一套可执行的排查顺序

如果以后再遇到类似错误，可以按下面的顺序快速处理：

先看堆栈关键词，判断是不是类型读错
查看表 schema
直接读 parquet 文件看真实 schema
按分区缩小范围
必要时关闭 vectorized reader 辅助定位
选择改表、重写分区或统一写入逻辑中的一种修复方式

这类问题在 Spark + Parquet 场景中非常常见，但本质上都不是“代码写错了”，而是：

数据一致性和元数据治理出了问题。

hnbian

https://www.hnbian.cn/posts/aea8ea3b.html

本博客所有文章除特別声明外，均采用 CC BY 4.0 许可协议。转载请注明来源 hnbian !

exception spark

$Scala Maven Plugin 编译报错：Cannot run program ... jre\bin\javac$ Scala Maven Plugin 编译报错：Cannot run program ... jre\bin\javac

2022-11-02 Scala

Scala

Spark 数据倾斜分析与解决思路

2022-08-08 spark

exception spark

Spark SQL 读取 Parquet 失败排查：Schema 不一致

一、问题现象

二、从原理理解这个错误

1. Parquet 的存储机制

2. Spark 是如何读取 Parquet 的

3. 为什么会报 getUTF8String -> decodeToBinary -> PlainLongDictionary

第一步：Spark 调用了 getUTF8String

第二步：Spark 进一步调用 decodeToBinary

第三步：底层真实对象却是 PlainLongDictionary

Schema 不一致

三、常见触发场景

场景 1：Hive 表结构与 Parquet 文件不一致

场景 2：同一张表不同分区类型不一致

场景 3：历史数据与新版本代码的字段类型变更

场景 4：直接写 Parquet 路径，再通过外部表读取

场景 5：多数据源汇总写入同一路径

四、如何定位问题

1. 先看表结构

2. 查看 Spark 侧 DataFrame schema

3. 直接读底层 Parquet 文件

4. 按分区缩小范围

5. 关闭向量化读取辅助定位

6. 必要时直接读单个文件

五、解决方案

方案一：修正表 schema（推荐）

方案二：重写异常分区

方案三：统一写入逻辑（根治）

方案四：临时规避（不推荐长期使用）

方案五：重建表或切换新路径

六、总结

最佳实践

一套可执行的排查顺序

3. 为什么会报 `getUTF8String -> decodeToBinary -> PlainLongDictionary`

第一步：Spark 调用了 `getUTF8String`

第二步：Spark 进一步调用 `decodeToBinary`

第三步：底层真实对象却是 `PlainLongDictionary`