Как мне загрузить файл на s3 с помощью Spark?

Question

Я установил спарк через pip install pyspark

Я использую следующий код для создания кадра данных из файла на s3.

from pyspark.sql import SparkSession

spark = SparkSession.builder \
            .config('spark.driver.extraClassPath', '/home/ubuntu/spark/jars/aws-java-sdk-1.11.335.jar:/home/ubuntu/spark/jars/hadoop-aws-2.8.4.jar') \
            .appName("cluster").getOrCreate()
df = spark.read.load('s3a://bucket/path/to/file')

Однако я получил ошибку:

--------------------------------------------------------------------------- Py4JJavaError Traceback (последний последний вызов) в () ----> 1 df =spark.read.load ('s3a: // bucket / path / to / file')

~ / miniconda3 / envs / аудитория / lib / python3.6 / site-packages / pyspark / sql / readwriter.py в загрузке (self, path, format, schema, ** options) 164 self.options (** options) 165 if isinstance (path, basestring): -> 166 возвращает self._df (self._jreader.load (path))) 167 elif путь не None: 168 если тип (путь)! = Список:

~ / miniconda3 / envs / аудитория / lib / python3.6 / site-packages / py4j / java_gateway.py в вызов (self, * args) 1158 answer = self.gateway_client.send_command (команда) 1159 return_value = get_return_value (-> 1160 ответ, self.gateway_client, self.target_id, self.name) 1161 1162 для temp_arg вtemp_args:

~ / miniconda3 / envs / аудитория / lib / python3.6 / site-packages / pyspark / sql / utils.py in deco (* a, ** kw) 61 def deco (* a, ** kw): 62 try:---> 63 возвращает f (* a, ** kw) 64 за исключением py4j.protocol.Py4JJavaError как e: 65 s = e.java_exception.toString ()

~ / miniconda3 / envs / аудитория / библиотека/python3.6/site-packages/py4j/protocol.py в get_return_value (answer, gateway_client, target_id, name) 318 вызывает Py4JJavaError (319 "Произошла ошибка при вызове {0} {1} {2}. \ n".-> 320 формат (target_id, ".", Name), значение) 321 else: 322 повысить Py4JError (

Py4JJavaError: Произошла ошибка при вызове o220.load.: Java.lang.NoClassDefFoundError: org /apache / hadoop / fs / StorageStatistics на java.lang.Class.forName0 (собственный метод) на java.lang.Class.forName (Class.java:348) на org.apache.hadoop.conf.Configuration.getClassByNameOrNull (Configuration.java): 2134) в org.apache.hadoop.conf.Configuration.getClassByName (Configuration.java:2099) в org.apache.hadoop.conf.Configuration.getClass (Configuration.java:2193) в org.apache.hadoop.fs.FileSystem.getFileSystemClass (FileSystem.java:2654) в org.apache.hadoop.fs.FileSystem.createFileSystem (FileSystem.java:2667) в org.apache.hadoop.fs.FileSystem.access $ 200 (FileSystem.java:94) вorg.apache.hadoop.fs.FileSystem $ Cache.getInternal (FileSystem.java:2703) в org.apache.hadoop.fs.FileSystem $ Cache.get (FileSystem.java:2685) в org.apache.hadoop.fs.FileSystem.get (FileSystem.java:373) в org.apache.hadoop.fs.Path.getFileSystem (Path.java:295) в org.apache.spark.sql.execution.streaming.FileStreamSink $ .hasMetadata (FileStreamSink.scala: 44) в org.apache.spark.sql.execution.datasources.DataSource.resolveRelation (DataSource.scala: 354) в org.apache.spark.sql.DataFrameReader.loadV1Source (DataFrameReader.scala: 239) в org.apache.spark.sql.DataFrameReader.load (DataFrameReader.scala: 227) в org.apache.spark.sql.DataFrameReader.load (DataFrameReader.scala: 174) по адресу sun.reflect.NativeMethodAccessorImpl.invoke0 (собственный метод) по адресу sun.reflect.NativeMethodAccessorImpl.invoke (NativeMethodAccessorImpl.java.Impor.Impor.Java: 43) на java.lang.reflect.Method.invoke (Method.java:498) на py4j.reflection.MethodInvoker.invoke (MethodInvoker.java:244) на py4j.reflection.ReflectionEngine.invoke (ReflectionEngine.java:35)) в py4j.Gateway.invoke (Gateway.java:282) в py4j.commands.AbstractCommand.invokeMethod (AbstractCommand.java:132)на py4j.commands.CallCommand.execute (CallCommand.java:79) на py4j.GatewayConnection.run (GatewayConnection.java:214) на java.lang.Thread.run (Thread.java:748). Причина: java.lang.ClassNotFoundException: org.apache.hadoop.fs.StorageStatistics на java.net.URLClassLoader.findClass (URLClassLoader.java:381) на java.lang.ClassLoader.loadClass (ClassLoader.java:424) на уровне sun.misc.Launco $.loadClass (Launcher.java:349) на java.lang.ClassLoader.loadClass (ClassLoader.java:357) ... еще 28

Если я изменю s3a на s3 или s3n, он запросит ключ доступа к aws.Однако я уже даю экземпляр ec2 AmazonS3FullAccess в IAM.

IllegalArgumentException: 'Идентификатор ключа доступа AWS и секретный ключ доступа должны быть указаны путем установки fs.s3.awsAccessKeyId и fs.s3.Свойства awsSecretAccessKey (соответственно). '

Любая помощь приветствуется.

Answer 1

Вам нужен способ предоставить свои учетные данные AWS сценарию.

Пример использования botocore, приведенный ниже, может оказаться слишком сложным, но избавляет вас от необходимости развертывать собственную конфигурацию AWS или парсер учетных данных.

Сначала

pip install botocore

Затем создайте сеанс и разрешите свои учетные данные вслепую.Порядок разрешения учетных данных задокументирован здесь

from pyspark.sql import SparkSession
import botocore.session

session = botocore.session.get_session()
credentials = session.get_credentials()

spark = (
    SparkSession
    .builder
    .config(
        'spark.driver.extraClassPath', 
        '/home/ubuntu/spark/jars/aws-java-sdk-1.11.335.jar:'
        '/home/ubuntu/spark/jars/hadoop-aws-2.8.4.jar')
    .config('fs.s3a.access.key', credentials.access_key)
    .config('fs.s3a.secret.key', credentials.secret_key)
    .appName("cluster")
    .getOrCreate()
)

df = spark.read.load('s3a://bucket/path/to/file')

РЕДАКТИРОВАТЬ

При использовании клиента файловой системы s3n свойства аутентификации такие же

.config('fs.s3n.awsAccessKeyId', credentials.access_key)
.config('fs.s3n.awsSecretAccessKey', credentials.secret_key)

Answer 2

Первая ошибка говорит вам, что Spark пытается загрузить класс org.apache.hadoop.fs.StorageStatistics.Не могли бы вы убедиться, что ваша версия Spark соответствует вашему JAR Hadoop?Обычно класс, который пытается загрузить Spark, был добавлен в этот коммит https://github.com/apache/hadoop/commit/687233f20d24c29041929dd0a99d963cec54b6df#diff-114b1833bd381e88382ade201dc692e8 и в отношении тегов выпуска, выпущенных первыми в 3.0.0.Поскольку вы используете Hadoop 2.8.4, обновление до 3.0.0 может стать решением.