PySpark 使用 IAM 角色访问 S3

Question

我想知道 PySpark 是否支持使用 IAM 角色进行 S3 访问。具体来说，我有一个业务限制，我必须承担 AWS 角色才能访问给定的存储桶。这在使用 boto 时很好（因为它是 API 的一部分），但我无法找到关于 PySpark 是否支持开箱即用的明确答案。

理想情况下，我希望能够在本地以独立模式运行时担任一个角色，并将我的 SparkContext 指向该 s3 路径。我已经看到非 IAM 调用通常会跟随：

spark_conf = SparkConf().setMaster('local[*]').setAppName('MyApp')
sc = SparkContext(conf=spark_conf)
rdd = sc.textFile('s3://<MY-ID>:<MY-KEY>@some-bucket/some-key')

是否存在用于提供 IAM 信息的类似方法？ ：

rdd = sc.textFile('s3://<MY-ID>:<MY-KEY>:<MY-SESSION>@some-bucket/some-key')

或

rdd = sc.textFile('s3://<ROLE-ARN>:<ROLE-SESSION-NAME>@some-bucket/some-key')

如果没有，使用 IAM 凭据的最佳做法是什么？有可能吗？

我正在使用 Python 1.7 和 PySpark 1.6.0

谢谢！

Answer 1

用于访问s3 的 IAM 角色仅由 s3a 支持，因为它使用的是 AWS SDK。

您需要将hadoop-aws JAR 和aws-java-sdk JAR（及其包中的第三方Jar）放入您的CLASSPATH。

hadoop-aws 链接。

aws-java-sdk链接。

然后在core-site.xml中设置这个：

<property>
    <name>fs.s3.impl</name>
    <value>org.apache.hadoop.fs.s3a.S3AFileSystem</value>
</property>
<property>
    <name>fs.s3a.impl</name>
    <value>org.apache.hadoop.fs.s3a.S3AFileSystem</value>
</property>

Answer 2

Hadoop 2.8+ 的 s3a 连接器通过新的凭据提供程序支持 IAM 角色；它不在 Hadoop 2.7 版本中。

要使用它，您需要更改凭据提供程序。

fs.s3a.aws.credentials.provider = org.apache.hadoop.fs.s3a.TemporaryAWSCredentialsProvider
fs.s3a.access.key = <your access key>
fs.s3a.secret.key = <session secret>
fs.s3a.session.token = <session token>

Hadoop 2.7 中的功能（默认启用）是获取AWS_ 环境变量。

如果您将AWS env vars 设置为本地系统和远程系统上的会话登录，那么它们应该会被接收。

我知道这很痛苦，但就 Hadoop 团队而言，Hadoop 2.7 于 2016 年年中发布，从那时起我们已经做了很多工作，我们不打算向后移植

Answer 3

Spark 支持对 S3 中的文件进行基于角色的访问，您只需要小心配置即可。具体来说，您需要：

• aws-java-sdk 和 hadoop-aws 的兼容版本。这是quite brittle，所以只有特定的组合才有效。
• 您必须使用S3AFileSystem，而不是NativeS3FileSystem。前者允许基于角色的访问，而后者只允许用户凭据。

要了解哪些组合有效，请访问 mvnrepository here 上的 hadoop-aws。点击hadoop-aws的版本，找到aws-java-sdk编译依赖的version。

要找出您正在使用的hadoop-aws 的版本，您可以在 PySpark 中执行：

sc._gateway.jvm.org.apache.hadoop.util.VersionInfo.getVersion()

sc 是 SparkContext

这对我有用：

import os
import pyspark
from pyspark import SparkContext
from pyspark.sql import SparkSession

os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages com.amazonaws:aws-java-sdk:1.7.4,org.apache.hadoop:hadoop-aws:2.7.1 pyspark-shell'

sc = SparkContext.getOrCreate()

hadoopConf = sc._jsc.hadoopConfiguration()
hadoopConf.set("fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem")

spark = SparkSession(sc)

df = spark.read.csv("s3a://mybucket/spark/iris/",header=True)
df.show()

是aws-java-sdk:1.7.4 和hadoop-aws:2.7.1 的特定组合使它起作用。 s3a 访问问题排查有很好的指导here

特别注意

随机更改 hadoop- 和 aws- JAR 以希望让问题“消失”或获得您想要的功能，不会导致您想要的结果。

这是一个包含更多信息的useful post。

这里有更多关于 java 库之间兼容性的useful information

我试图让它在jupyter pyspark notebook 中工作。请注意，aws-hadoop 版本必须与 Dockerfile 中的 hadoop 安装相匹配，即here。

Answer 4

您可以尝试Locally reading S3 files through Spark (or better: pyspark) 中的方法。

但是，我在 Bash 中设置环境变量（AWS_ACCESS_KEY_ID 等）的运气更好……pyspark 会自动为您的会话选择这些。

Answer 5

经过更多研究，我确信这一点尚未得到支持，正如 here 所证明的那样。

其他人建议采用更手动的方法（请参阅this blog post），该方法建议使用 boto 列出 s3 键，然后使用 Spark 并行化该列表以读取每个对象。

这里的问题（我还没有看到他们自己如何解决它）是从存储桶中列出返回的 s3 对象不可序列化/可腌制（请记住：建议给出这些对象让工人通过 map 或 flatMap 在独立的进程中读取）。进一步的问题是boto s3客户端本身不可序列化（我认为这是合理的）。

我们剩下的唯一选择是重新创建假定角色的 s3 客户端每个文件，这在特定点之后不是最优或不可行的。

如果有人在这种推理或替代解决方案/方法中发现任何缺陷，我很乐意听到。