如何修复PySpark中的java.lang.ClassNotFoundException: PathOutputCommitProtocol错误

运行环境 #

Java Home   /Library/Java/JavaVirtualMachines/jdk-11.0.16.1.jdk/Contents/Home
Java Version    11.0.16.1 (Oracle Corporation)
Scala Version   version 2.12.15
Spark Version.         spark-3.3.1
Python 3.9

执行代码 #

尝试将PySpark DataFrame写入本地CSV文件：

df.write.csv("/Users/data/data.csv")

报错信息 #

Traceback (most recent call last):
  File "/Applications/PyCharm.app/Contents/plugins/python/helpers/pydev/_pydevd_bundle/pydevd_exec2.py", line 3, in Exec
    exec(exp, global_vars, local_vars)
  File "<input>", line 1, in <module>
  File "/Users/mambaforge-pypy3/envs/lib/python3.9/site-packages/pyspark/sql/readwriter.py", line 1240, in csv
    self._jwrite.csv(path)
  File "/Users/mambaforge-pypy3/envs/lib/python3.9/site-packages/py4j/java_gateway.py", line 1321, in __call__
    return_value = get_return_value(
  File "/Users/mambaforge-pypy3/envs/lib/python3.9/site-packages/pyspark/sql/utils.py", line 190, in deco
    return f(*a, **kw)
  File "/Users/mambaforge-pypy3/envs/lib/python3.9/site-packages/py4j/protocol.py", line 326, in get_return_value
    raise Py4JJavaError(
py4j.protocol.Py4JJavaError: An error occurred while calling o747.csv.
: java.lang.ClassNotFoundException: org.apache.spark.internal.io.cloud.PathOutputCommitProtocol

当前Spark配置 #

spark_conf = SparkConf()
        spark_conf.setAll(parameters.items())
        spark_conf.set('spark.jars.packages', 'org.apache.hadoop:hadoop-aws:3.3.4')
        spark_conf.set('spark.hadoop.fs.s3.aws.credentials.provider',
                       'org.apache.hadoop.fs.s3.TemporaryAWSCredentialsProvider')
        spark_conf.set('spark.hadoop.fs.s3.access.key', os.environ.get('AWS_ACCESS_KEY_ID'))
        spark_conf.set('spark.hadoop.fs.s3.secret.key', os.environ.get('AWS_SECRET_ACCESS_KEY'))
        spark_conf.set('spark.hadoop.fs.s3a.bucket.all.committer.magic.enabled', 'true')
        spark_conf.set("com.amazonaws.services.s3.enableV4", "true")
        spark_conf.set("fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem")
        spark_conf.set("fs.s3a.aws.credentials.provider",
                       "com.amazonaws.auth.InstanceProfileCredentialsProvider,com.amazonaws.auth.DefaultAWSCredentialsProviderChain")
        spark_conf.set("fs.AbstractFileSystem.s3a.impl", "org.apache.hadoop.fs.s3a.S3A")
        spark_conf.set("hadoop.fs.s3a.path.style.access", "true")
        spark_conf.set("hadoop.fs.s3a.fast.upload", "true")
        spark_conf.set("hadoop.fs.s3a.fast.upload.buffer", "bytebuffer")
        spark_conf.set("fs.s3a.path.style.access", "true")
        spark_conf.set("fs.s3a.multipart.size", "128M")
        spark_conf.set("fs.s3a.fast.upload.active.blocks", "4")
        spark_conf.set("fs.s3a.committer.name", "partitioned")
        spark_conf.set("spark.hadoop.fs.s3a.committer.name", "directory")
        spark_conf.set("spark.sql.sources.commitProtocolClass",
                       "org.apache.spark.internal.io.cloud.PathOutputCommitProtocol")
        spark_conf.set("spark.sql.parquet.output.committer.class",
                       "org.apache.spark.internal.io.cloud.BindingParquetOutputCommitter")
        spark_conf.set("spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version", "1")

问题根源 #

你配置了针对S3云存储的输出提交协议类org.apache.spark.internal.io.cloud.PathOutputCommitProtocol，但该类属于spark-hadoop-cloud模块，当前仅引入了hadoop-aws依赖，缺少对应模块；同时你写入的是本地路径，不需要启用云存储的提交协议，导致本地文件系统尝试加载不存在的类。

具体解决方法 #

方法1：写入本地时临时覆盖提交协议

因为写入本地路径不需要S3的提交协议，直接在写入时指定本地默认的提交协议类：

df.write.option("spark.sql.sources.commitProtocolClass", "org.apache.spark.sql.execution.datasources.FileOutputCommitProtocol")\
        .csv("/Users/data/data.csv")

方法2：补充缺失的依赖包

如果需要同时支持本地和S3写入，在spark.jars.packages中添加对应版本的spark-hadoop-cloud包（版本需和Spark 3.3.1匹配）：

spark_conf.set('spark.jars.packages', 'org.apache.hadoop:hadoop-aws:3.3.4,org.apache.spark:spark-hadoop-cloud_2.12:3.3.1')

添加后Spark会自动下载包含PathOutputCommitProtocol类的依赖包。

方法3：清理不必要的S3配置

如果日常操作以本地为主，可以移除SparkConf中针对S3的提交协议配置：

# 移除这两行配置
# spark_conf.set("spark.sql.sources.commitProtocolClass", "org.apache.spark.internal.io.cloud.PathOutputCommitProtocol")
# spark_conf.set("spark.sql.parquet.output.committer.class", "org.apache.spark.internal.io.cloud.BindingParquetOutputCommitter")

仅在需要写入S3时再添加这些配置。

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