Spark写入Delta格式至AWS S3数据湖报错排查求助

问题 #

目标是将Delta格式数据上传至AWS S3数据湖，此前已实现向该数据湖写入Parquet文件，本地可正常写入Delta格式，但写入Delta至S3时报错。同时添加Delta相关配置后，Parquet写入也失败，移除Delta配置后Parquet恢复正常。

环境：Spark 3.4.1、Scala 2.12、Delta-Spark 2.4.0

代码 #

import findspark
from pyspark.sql import SparkSession
import pandas as pd

findspark.find()
findspark.init()

import boto3

# Initialize a session using the AWS SDK for Python (boto3)
session = boto3.Session(profile_name='default')

# Get the AWS credentials
credentials = session.get_credentials()

spark = SparkSession.builder.appName('Session2').config('spark.master', 'local[4]') \
                            .config("spark.hadoop.fs.s3a.access.key", credentials.access_key) \
                            .config("spark.hadoop.fs.s3a.secret.key", credentials.secret_key) \
                            .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
                            .config("spark.jars.packages", "org.apache.hadoop:hadoop-aws:3.3.6,org.apache.hadoop:hadoop-common:3.3.6,io.delta:delta-core_2.12:2.4.0") \
                            .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
                            .config("spark.sql.legacy.parquet.int96RebaseModeInWrite","CORRECTED") \
                            .config("spark.hadoop.fs.s3a.impl", "org.apache.spark.sql.delta.storage.S3SingleDriverLogStore") \
                            .config("spark.delta.logStore.class", "org.apache.spark.sql.delta.storage.S3SingleDriverLogStore") \
                            .getOrCreate()      

df = ... # some dataframe

df.write.format("delta").option("overwriteSchema", "true").mode("overwrite").option("mergeSchema", "true").save('s3a://bucket/dts/deltatable/')

Delta写入S3报错信息 #

"name": "Py4JJavaError",
    "message": "An error occurred while calling o116.save.\n: java.util.concurrent.ExecutionException: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 25.0 failed 1 times, most recent failure: Lost task 0.0 in stage 25.0 (TID 172) (host.docker.internal executor driver): java.lang.NoSuchMethodError: 'java.lang.Object org.apache.hadoop.fs.statistics.impl.IOStatisticsBinding.invokeTrackingDuration(org.apache.hadoop.fs.statistics.DurationTracker, org.apache.hadoop.util.functional.CallableRaisingIOE)'\r\n\tat org.apache.hadoop.fs.s3a.Invoker.onceTrackingDuration(Invoker.java:147)\r\n\tat org.apache.hadoop.fs.s3a.S3AInputStream.reopen(S3AInputStream.java:282)\r\n\tat org.apache.hadoop.fs.s3a.S3AInputStream.lambda$lazySeek$1(S3AInputStream.java:435)\r\n\tat org.apache.hadoop.fs.s3a.Invoker.lambda$maybeRetry$3(Invoker.java:284)\r\n\tat org.apache.hadoop.fs.s3a.Invoker.once(Invoker.java:122)\r\n\tat org.apache.hadoop.fs.s3a.Invoker.lambda$maybeRetry$5(Invoker.java:408)\r\n\tat org.apache.hadoop.fs.s3a.Invoker.retryUntranslated(Invoker.java:468)\r\n\tat org.apache.hadoop.fs.s3a.Invoker.maybeRetry(Invoker.java:404).....

Parquet写入S3报错信息（添加Delta配置后） #

Py4JJavaError: An error occurred while calling o130.parquet.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 27.0 failed 1 times, most recent failure: Lost task 0.0 in stage 27.0 (TID 177) (host.docker.internal executor driver): org.apache.spark.SparkException: Python worker failed to connect back.
    at org.apache.spark.api.python.PythonWorkerFactory.createSimpleWorker(PythonWorkerFactory.scala:192)
    at org.apache.spark.api.python.PythonWorkerFactory.create(PythonWorkerFactory.scala:109)
    at org.apache.spark.SparkEnv.createPythonWorker(SparkEnv.scala:124)
    at org.apache.spark.api.python.BasePythonRunner.compute(PythonRunner.scala:166)
    at org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:65)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:364)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:328)
    at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:364)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:328)
    at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:364)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:328)
    at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:364)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:328)
    at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:364)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:328)

解决方法 #

1. 修复Hadoop依赖版本冲突 #

NoSuchMethodError是因为Hadoop版本不兼容导致的方法缺失。Spark 3.4.1默认依赖Hadoop 3.3.4，你指定的hadoop-aws:3.3.6、hadoop-common:3.3.6与Spark自带组件版本不一致，引发依赖冲突。

调整spark.jars.packages，使用匹配版本：

.config("spark.jars.packages", "org.apache.hadoop:hadoop-aws:3.3.4,io.delta:delta-core_2.12:2.4.0")

无需显式指定hadoop-common，Spark会自动引入匹配版本，避免重复依赖冲突。

2. 修正S3文件系统配置 #

你错误地将spark.hadoop.fs.s3a.impl设置为Delta的LogStore类，正确配置应为：

.config("spark.hadoop.fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem")

spark.delta.logStore.class保持原配置即可，它负责Delta的日志存储：

.config("spark.delta.logStore.class", "org.apache.spark.sql.delta.storage.S3SingleDriverLogStore")

3. 解决Python Worker连接问题 #

Python Worker无法连接通常是依赖冲突导致JVM与Python进程通信异常，修复上述配置后该问题会自动解决。若仍出现问题，可添加以下配置调整内存和超时：

.config("spark.python.worker.memory", "4g")
.config("spark.executorEnv.PYTHONPATH", "")

4. 验证最终配置 #

调整后的SparkSession构建代码：

spark = SparkSession.builder.appName('Session2').config('spark.master', 'local[4]') \
                            .config("spark.hadoop.fs.s3a.access.key", credentials.access_key) \
                            .config("spark.hadoop.fs.s3a.secret.key", credentials.secret_key) \
                            .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
                            .config("spark.jars.packages", "org.apache.hadoop:hadoop-aws:3.3.4,io.delta:delta-core_2.12:2.4.0") \
                            .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
                            .config("spark.sql.legacy.parquet.int96RebaseModeInWrite","CORRECTED") \
                            .config("spark.hadoop.fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem") \
                            .config("spark.delta.logStore.class", "org.apache.spark.sql.delta.storage.S3SingleDriverLogStore") \
                            .getOrCreate()

5. 额外注意事项 #

• S3SingleDriverLogStore仅适用于本地模式或单节点集群，生产环境建议使用S3DynamoDBLogStore保证多节点日志一致性。
• 确保AWS权限配置正确，S3 Bucket需允许读写操作，IAM凭证具备s3:PutObject、s3:GetObject等权限。

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