Чтение данных из MinIO с помощью pyspark - PullRequest
Новая
       102

Чтение данных из MinIO с помощью pyspark

0 голосов
/ 21 марта 2020

Я пытаюсь прочитать данные из MinIO с pyspark с этим кодом: 

SCHEMA = ... # data schema

spark = SparkSession.builder \
                    .config('spark.jars.packages', 'org.apache.hadoop:hadoop-common:2.7.1,' \
                                                   'org.apache.hadoop:hadoop-aws:2.7.1,' \
                                                   'com.amazonaws:aws-java-sdk:1.7.4,' \
                                                   'io.minio:spark-select_2.11:2.1') \
                    .config('spark.hadoop.fs.s3a.endpoint', 'http://minio:9000') \
                    .config('spark.hadoop.fs.s3a.access.key', 'minioadmin') \
                    .config('spark.hadoop.fs.s3a.secret.key', 'minioadmin') \
                    .config('spark.hadoop.fs.s3a.path.style.access', 'true') \
                    .config('spark.hadoop.fs.s3a.impl', 'org.apache.hadoop.fs.s3a.S3AFileSystem') \
                    .getOrCreate()

spark.read.format("minioSelectJSON").schema(SCHEMA).load("s3://bucket/path/to/data.json)

Но я застрял в следующей ошибке: 

py4j.protocol.Py4JJavaError: An error occurred while calling o62.json.
: java.lang.NoSuchMethodError: com.amazonaws.services.s3.transfer.TransferManager.<init>(Lcom/amazonaws/services/s3/AmazonS3;Ljava/util/concurrent/ThreadPoolExecutor;)V
        at org.apache.hadoop.fs.s3a.S3AFileSystem.initialize(S3AFileSystem.java:287)
        at org.apache.hadoop.fs.FileSystem.createFileSystem(FileSystem.java:2669)
        at org.apache.hadoop.fs.FileSystem.access$200(FileSystem.java:94)
        at org.apache.hadoop.fs.FileSystem$Cache.getInternal(FileSystem.java:2703)
        at org.apache.hadoop.fs.FileSystem$Cache.get(FileSystem.java:2685)
        at org.apache.hadoop.fs.FileSystem.get(FileSystem.java:373)
        at org.apache.hadoop.fs.Path.getFileSystem(Path.java:295)
        at org.apache.spark.sql.execution.streaming.FileStreamSink$.hasMetadata(FileStreamSink.scala:45)
        at org.apache.spark.sql.execution.datasources.DataSource.resolveRelation(DataSource.scala:332)
        at org.apache.spark.sql.DataFrameReader.loadV1Source(DataFrameReader.scala:223)
        at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:211)
        at org.apache.spark.sql.DataFrameReader.json(DataFrameReader.scala:392)
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.lang.reflect.Method.invoke(Method.java:498)
        at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
        at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
        at py4j.Gateway.invoke(Gateway.java:282)
        at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
        at py4j.commands.CallCommand.execute(CallCommand.java:79)
        at py4j.GatewayConnection.run(GatewayConnection.java:238)
        at java.lang.Thread.run(Thread.java:745)

После прочтения аналогичных вопросов это Кажется, что проблема заключается в использовании несовместимых версий используемых пакетов, но я не смог найти комбинацию версий, которая бы работала. У вас есть идеи, как это исправить?

Спасибо за любую помощь.

...