Pyspark, Hadoop и S3: java.lang.NoSuchMethodError: org.apache.hadoop.fs.s3a.Listing$FileStatusListingIterator ⇐ Python

[Anonymous]

Я столкнулся с проблемами совместимости, связанными с работой delta-spark с S3 прямо из коробки, и хотел получить совет. Я пробовал десятки комбинаций версий между Spark, delta-spark, Hadoop-aws и aws-java-sdk-bundle, но продолжал получать некоторые варианты NoSuchMethodError, которые явно указывают на ошибку совместимости.< /p>
Я пробовал использовать самые последние версии Spark (3.5.3) и delta-spark (3.2.1), но теперь я пробую явный пример, указанный на их веб-сайте.
Согласно: https://docs.delta.io/latest/delta-stor ... le-cluster
Они говорят:

Используйте следующую команду для запуска оболочки Spark с поддержкой Delta Lake и S3 (при условии, что вы используете Spark 3.5.0, предварительно созданный для Hadoop 3.3.4):

и использует --packages io.delta:delta-spark_2.12:3.1.0,org.apache.hadoop:hadoop-aws:3.3. 4
И согласно их матрице совместимости (https://docs.delta.io/latest/releases.html), 3.1.0 работает с 3.5.0. Поэтому я переустановил пакеты с pyspark 3.5.0 и delta-spark 3.1.0.
Мой код выглядит следующим образом:

Код: Выделить всё

builder = pyspark.sql.SparkSession.builder.appName(app_name) \
.config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
.config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
.config("spark.hadoop.fs.s3a.aws.credentials.provider", "org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider") \
.config("spark.hadoop.fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem") \
.config("spark.hadoop.fs.s3a.access.key", aws_access_key_id) \
.config("spark.hadoop.fs.s3a.secret.key", secret_access_key) \
.config("spark.hadoop.fs.s3a.endpoint", f"s3.{region_name}.amazonaws.com") \
.config("spark.hadoop.delta.enableFastS3AListFrom", True)

extra_packages = ["org.apache.hadoop:hadoop-aws:3.3.4,com.amazonaws:aws-java-sdk-bundle:1.12.262"]
spark = delta.configure_spark_with_delta_pip(
builder,
extra_packages=extra_packages,
).getOrCreate()
data = {
'id': [1, 2, 3],
'name': ['Alice', 'Bob', 'Charlie'],
}
df = spark.createDataFrame(pd.DataFrame(data))
deltalake_table_spark = f"s3a://{bucket}/test_folder/spark_deltalake_table"
df.write.format("delta").save(deltalake_table_spark)

Где находится часть вывода ошибки:

Код: Выделить всё

Traceback (most recent call last):
File "...", line 183, in test_delta_spark
df.write.format("delta").save(deltalake_table_spark)
File ".../lib/python3.10/site-packages/pyspark/sql/readwriter.py", line 1463, in save
self._jwrite.save(path)
File ".../lib/python3.10/site-packages/py4j/java_gateway.py", line 1322, in __call__
return_value = get_return_value(
File ".../lib/python3.10/site-packages/pyspark/errors/exceptions/captured.py", line 179, in deco
return f(*a, **kw)
File ".../lib/python3.10/site-packages/py4j/protocol.py", line 326, in get_return_value
raise Py4JJavaError(
py4j.protocol.Py4JJavaError: An error occurred while calling o69.save.
: com.google.common.util.concurrent.ExecutionError: java.lang.NoSuchMethodError: 'org.apache.hadoop.fs.s3a.Listing$FileStatusListingIterator org.apache.hadoop.fs.s3a.Listing.createFileStatusListingIterator(org.apache.hadoop.fs.Path, org.apache.hadoop.fs.s3a.S3ListRequest, org.apache.hadoop.fs.PathFilter, org.apache.hadoop.fs.s3a.Listing$FileStatusAcceptor)'
at com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2261)

И для проверки

Код: Выделить всё

spark._sc._jvm.org.apache.hadoop.util.VersionInfo.getVersion()
Out[18]: '3.3.4'
spark.version
Out[19]: '3.5.0'

Что касается моей Java-версии:

Код: Выделить всё

java --version
openjdk 11.0.24 2024-07-16
OpenJDK Runtime Environment Homebrew (build 11.0.24+0)

Проверяя вывод журнала ошибок, можно увидеть, что установлены правильные версии:

Код: Выделить всё

:: modules in use:
com.amazonaws#aws-java-sdk-bundle;1.12.262 from central in [default]
io.delta#delta-spark_2.12;3.1.0 from central in [default]
io.delta#delta-storage;3.1.0 from central in [default]
org.antlr#antlr4-runtime;4.9.3 from central in [default]
org.apache.hadoop#hadoop-aws;3.3.4 from central in [default]
org.wildfly.openssl#wildfly-openssl;1.0.7.Final from central in [default]

Кажется, все синхронизировано. 3.3.4 Hadoop и Hadoop-aws, 3.1.0 delta-spark, 3.5.0 pyspark, и я пробовал оставить aws-java-sdk-bundle или изменить его в зависимости от совместимости версий.
Другие вещи, которые я пробовал:

• Вместо того, чтобы определять конфигурацию Hadoop в сборщике, я пытался сделать это после getOrCreate с помощью spark._jsc.hadoopConfiguration().set(...)
• Вместо configure_spark_with_delta_pip я вручную ввел дельту -spark jar-пакет в конфигурации с .config("spark.jars.packages", "io.delta:delta-spark_2.12:3.1.0,org.apache.hadoop:hadoop-aws:3.3. 4,com.amazonaws:aws-java-sdk-bundle:1.12.262") в определении компоновщика.

Я Из вариантов и искал много других тем, но мне не удалось запустить это. Может ли кто-нибудь подтвердить, действительно ли это проблема совместимости или мне не хватает чего-то еще? Спасибо.

Подробнее здесь: https://stackoverflow.com/questions/790 ... p-fs-s3a-l