如何解决PySpark中的ANTLR Runtime版本不匹配错误？

问题场景 #

运行以下读取CSV的PySpark代码时：

from pyspark.sql import SparkSession

spark = (SparkSession.builder.appName("4-1")      
         .getOrCreate())

csv_file = "departuredelays.csv"

schema = "`date` STRING, `delay` INT, `distance` INT, `origin` STRING, `destination` STRING"
df = spark.read.csv(csv_file, schema, inferSchema=True, header=True)

触发ANTLR版本不匹配错误，核心报错及提示如下：

An error occurred while calling o28.schema.
: java.lang.ExceptionInInitializerError
at org.apache.spark.sql.catalyst.parser.AbstractSqlParser.parse(ParseDriver.scala:107)
...
Caused by: java.io.InvalidClassException: org.antlr.v4.runtime.atn.ATN; Could not deserialize ATN with version 3 (expected 4).
... 19 more

To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
ANTLR Tool version 4.8 used for code generation does not match the current runtime version 4.12.0
ANTLR Runtime version 4.8 used for parser compilation does not match the current runtime version 4.12.0

解决方案 #

1. 统一Python环境的ANTLR版本 #

Spark 3.x系列默认依赖ANTLR 4.8，若环境中存在高版本的antlr4-python3-runtime会引发冲突，执行以下命令统一版本：

pip uninstall antlr4-python3-runtime -y
pip install antlr4-python3-runtime==4.8

2. 清理Java环境的依赖冲突 #

检查运行环境的CLASSPATH，确保没有混入ANTLR 4.12.0的jar包。若用Maven/Gradle管理项目，需强制指定ANTLR 4.8版本并排除冲突依赖：

• Maven配置示例：

<dependency>
    <groupId>org.antlr</groupId>
    <artifactId>antlr4-runtime</artifactId>
    <version>4.8</version>
    <scope>runtime</scope>
</dependency>

• Gradle配置示例：

dependencies {
    implementation('org.apache.spark:spark-sql_2.12:3.3.0') {
        exclude group: 'org.antlr', module: 'antlr4-runtime'
    }
    implementation 'org.antlr:antlr4-runtime:4.8'
}

3. 代码临时修复（移除冗余配置） #

代码中同时指定手动schema和inferSchema=True属于冗余配置，且schema推断会触发ANTLR解析。直接移除inferSchema=True即可规避错误：

from pyspark.sql import SparkSession

spark = (SparkSession.builder.appName("4-1")      
         .getOrCreate())

csv_file = "departuredelays.csv"

schema = "`date` STRING, `delay` INT, `distance` INT, `origin` STRING, `destination` STRING"
df = spark.read.csv(csv_file, schema=schema, header=True)

内容的提问来源于stack exchange，提问作者Tavakoli