Ответ 1
Документация имеет формат: http://wiki.apache.org/hadoop/AmazonS3
s3n://ID:[email protected]/Path
Я передаю входные и выходные папки в качестве параметров для mapreduce программы подсчета слов с веб-страницы.
Ошибка ниже:
Состояние HTTP 500 - обработка запроса не выполнена; вложенное исключение java.lang.IllegalArgumentException: идентификатор ключа доступа AWS и секрет Ключ доступа должен быть указан как имя пользователя или пароль (соответственно) URL-адреса s3n или установкой fs.s3n.awsAccessKeyId или fs.s3n.awsSecretAccessKey(соответственно).
Документация имеет формат: http://wiki.apache.org/hadoop/AmazonS3
s3n://ID:[email protected]/Path
Я предлагаю вам использовать это:
hadoop distcp \
-Dfs.s3n.awsAccessKeyId=<your_access_id> \
-Dfs.s3n.awsSecretAccessKey=<your_access_key> \
s3n://origin hdfs://destinations
Он также работает в качестве обходного пути для появления косой черты в ключе. Параметры с идентификатором и ключом доступа должны быть поставлены точно в этом порядке: после disctcp и перед началом
Передача в учетных данных AWS как части URL-адреса Amazon s3n обычно не рекомендуется, безопасность. Особенно, если этот код переносится на службу хранения репозитория (например, github). В идеале установите свои учетные данные в файле conf/core-site.xml как:
<configuration>
<property>
<name>fs.s3n.awsAccessKeyId</name>
<value>XXXXXX</value>
</property>
<property>
<name>fs.s3n.awsSecretAccessKey</name>
<value>XXXXXX</value>
</property>
</configuration>
или переустановите awscli на вашем компьютере.
pip install awscli
Для начинающих pyspark:
Загрузите флягу с https://mvnrepository.com/artifact/org.apache.hadoop/hadoop-aws
Поместите это в папку JAR
Тогда ты можешь
core-site.xml
export AWS_ACCESS_KEY_ID=<access-key>
export AWS_SECRET_ACCESS_KEY=<secret-key>
<configuration>
<property>
<name>fs.s3n.impl</name>
<value>org.apache.hadoop.fs.s3native.NativeS3FileSystem</value>
</property>
<property>
<name>fs.s3a.impl</name>
<value>org.apache.hadoop.fs.s3a.S3AFileSystem</value>
</property>
<property>
<name>fs.s3.impl</name>
<value>org.apache.hadoop.fs.s3.S3FileSystem</value>
</property>
</configuration>
sc._jsc.hadoopConfiguration().set("fs.s3.awsAccessKeyId", access_key)
sc._jsc.hadoopConfiguration().set("fs.s3n.awsAccessKeyId", access_key)
sc._jsc.hadoopConfiguration().set("fs.s3a.access.key", access_key)
sc._jsc.hadoopConfiguration().set("fs.s3.awsSecretAccessKey", secret_key)
sc._jsc.hadoopConfiguration().set("fs.s3n.awsSecretAccessKey", secret_key)
sc._jsc.hadoopConfiguration().set("fs.s3a.secret.key", secret_key)
sc._jsc.hadoopConfiguration().set("fs.s3n.impl", "org.apache.hadoop.fs.s3native.NativeS3FileSystem")
sc._jsc.hadoopConfiguration().set("fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem")
sc._jsc.hadoopConfiguration().set("fs.s3.impl", "org.apache.hadoop.fs.s3.S3FileSystem")
import sys
from random import random
from operator import add
from pyspark.sql import SparkSession
from pyspark.conf import SparkConf
if __name__ == "__main__":
"""
Usage: S3 sample
"""
access_key = '<access-key>'
secret_key = '<secret-key>'
spark = SparkSession\
.builder\
.appName("Demo")\
.getOrCreate()
sc = spark.sparkContext
# remove this block if use core-site.xml and env variable
sc._jsc.hadoopConfiguration().set("fs.s3.awsAccessKeyId", access_key)
sc._jsc.hadoopConfiguration().set("fs.s3n.awsAccessKeyId", access_key)
sc._jsc.hadoopConfiguration().set("fs.s3a.access.key", access_key)
sc._jsc.hadoopConfiguration().set("fs.s3.awsSecretAccessKey", secret_key)
sc._jsc.hadoopConfiguration().set("fs.s3n.awsSecretAccessKey", secret_key)
sc._jsc.hadoopConfiguration().set("fs.s3a.secret.key", secret_key)
sc._jsc.hadoopConfiguration().set("fs.s3n.impl", "org.apache.hadoop.fs.s3native.NativeS3FileSystem")
sc._jsc.hadoopConfiguration().set("fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem")
sc._jsc.hadoopConfiguration().set("fs.s3.impl", "org.apache.hadoop.fs.s3.S3FileSystem")
# fetch from s3, returns RDD
csv_rdd = spark.sparkContext.textFile("s3n://<bucket-name>/path/to/file.csv")
c = csv_rdd.count()
print("~~~~~~~~~~~~~~~~~~~~~count~~~~~~~~~~~~~~~~~~~~~")
print(c)
spark.stop()