Spark+S3+IAM角色

Question

我正尝试使用IAM角色从spark读取s3桶中的csv文件，但在MultiObjectDeleteException上获得NoClassDefoundErr

我安装了没有hadoop的Spark 2.4.4，安装了hadoop 3.2.1以及hadoop-aws-3.2.1.jar和aws-java-sdk-1.11.655.jar。我必须安装一个没有hadoop的spark版本，因为作为spark构建一部分的hadoop jars是来自2016年的2.7.3版本。

sc.hadoopConfiguration.set("fs.s3a.credentialsType", "AssumeRole")
sc.hadoopConfiguration.set("fs.s3a.assumed.role.arn", "arn:aws:iam::[ROLE]")
val myRDD = sc.textFile("s3a://test_bucket/names.csv")
myRDD.count()

附加到角色的“我的IAM策略”具有以下内容

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "VisualEditor0",
            "Effect": "Allow",
            "Action": [
                "s3:PutAccountPublicAccessBlock",
                "s3:GetAccountPublicAccessBlock",
                "s3:ListAllMyBuckets",
                "s3:ListJobs",
                "s3:CreateJob",
                "s3:HeadBucket"
            ],
            "Resource": "*"
        },
        {
            "Sid": "VisualEditor1",
            "Effect": "Allow",
            "Action": "s3:*",
            "Resource": "arn:aws:s3:::test_bucket"
        }
    ]
}

我甚至尝试过sc.hadoopconfiguration.set（“fs.s3a.multiobjectdelete.enable”,“false”），但错误如下：

java.lang.NoClassDefFoundError: com/amazonaws/services/s3/model/MultiObjectDeleteException
  at java.lang.Class.forName0(Native Method)
  at java.lang.Class.forName(Class.java:348)
  at org.apache.hadoop.conf.Configuration.getClassByNameOrNull(Configuration.java:2575)
  at org.apache.hadoop.conf.Configuration.getClassByName(Configuration.java:2540)
  at org.apache.hadoop.conf.Configuration.getClass(Configuration.java:2636)
  at org.apache.hadoop.fs.FileSystem.getFileSystemClass(FileSystem.java:3269)
  at org.apache.hadoop.fs.FileSystem.createFileSystem(FileSystem.java:3301)
  at org.apache.hadoop.fs.FileSystem.access$200(FileSystem.java:124)
  at org.apache.hadoop.fs.FileSystem$Cache.getInternal(FileSystem.java:3352)
  at org.apache.hadoop.fs.FileSystem$Cache.get(FileSystem.java:3320)
  at org.apache.hadoop.fs.FileSystem.get(FileSystem.java:479)
  at org.apache.hadoop.fs.Path.getFileSystem(Path.java:365)
  at org.apache.hadoop.mapred.FileInputFormat.singleThreadedListStatus(FileInputFormat.java:268)
  at org.apache.hadoop.mapred.FileInputFormat.listStatus(FileInputFormat.java:239)
  at org.apache.hadoop.mapred.FileInputFormat.getSplits(FileInputFormat.java:325)
  at org.apache.spark.rdd.HadoopRDD.getPartitions(HadoopRDD.scala:204)
  at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:253)
  at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:251)
  at scala.Option.getOrElse(Option.scala:121)
  at org.apache.spark.rdd.RDD.partitions(RDD.scala:251)
  at org.apache.spark.rdd.MapPartitionsRDD.getPartitions(MapPartitionsRDD.scala:49)
  at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:253)
  at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:251)
  at scala.Option.getOrElse(Option.scala:121)
  at org.apache.spark.rdd.RDD.partitions(RDD.scala:251)
  at org.apache.spark.SparkContext.runJob(SparkContext.scala:2126)
  at org.apache.spark.rdd.RDD.count(RDD.scala:1168)
  ... 49 elided
Caused by: java.lang.ClassNotFoundException: com.amazonaws.services.s3.model.MultiObjectDeleteException
  at java.net.URLClassLoader.findClass(URLClassLoader.java:382)
  at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
  at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:349)
  at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
  ... 76 more

Answer 1

没有选项fs.s3a.credentialstype；对于s3a,所有东西都是小写的,这有助于调试这些东西,

关于假定角色凭据的文档涵盖了所需的权限https://hadoop.apache.org/docs/r3.1.0/hadoop-aws/tools/hadoop-aws/assumed_roles.html

这在Hadoop3.2上的工作方式是，必须具有完全权限的调用，然后s3a连接器调用STS AssuseRole，以在给定角色中创建一些短期会话凭据。在EC2中，VM没有调用AssuseRole的权限（它们已经在一个角色中运行），所以您必须使用创建VM时使用的内容。

现在，使用s3a假定的角色来查看策略允许的角色。

Answer 2

上述问题与IAM政策有关。它没有查看文件的策略“/*”是必需的。

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "s3:GetBucketLocation",
                "s3:ListBucket"
            ],
            "Resource": [
                "arn:aws:s3:::test_bucket"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "s3:PutObject",
                "s3:PutObjectAcl",
                "s3:GetObject",
                "s3:DeleteObject"
            ],
            "Resource": [
                "arn:aws:s3:::test_bucket/*"
            ]
        }
    ]
}

您创建的角色将具有上述IAM策略。该角色将附加到EC2实例（主和从AWS EC2实例），这是非常关键的，因为Spark将恢复分配给EC2实例的角色。因此，因为EC2被分配了角色，所以您不需要在scala代码中指定角色。您需要做的只是编写以下Scala代码来读取一个文件，该文件将恢复分配给EC2实例的角色。

val myRDD = sc.textFile("s3a://test_bucket/test.csv")
myRDD.count()

hadoop-3.2.1.tar.gz在/opt/hadoop/share/hadoop/tools/lib中同时拥有hadoop-aws-3.2.1.jar和aws-java-sdk-bundle-1.11.375.jar

在这里，您希望确保定义了一个spark-env.sh指向正确的jar目录，以便spark加载类路径中的jar。

cp /opt/spark/conf/spark-env.sh.template /opt/spark/conf/spark-env.sh

export SPARK_DIST_CLASSPATH=/opt/spark/jars:/opt/hadoop/etc/hadoop:/opt/hadoop/share/hadoop/common/lib/*:/opt/hadoop/share/hadoop/common/*:/opt/hadoop/share/hadoop/hdfs:/opt/hadoop/share/hadoop/hdfs/lib/*:/opt/hadoop/share/hadoop/hdfs/*:/opt/hadoop/share/hadoop/yarn/lib/*:/opt/hadoop/share/hadoop/yarn/*:/opt/hadoop/share/hadoop/mapreduce/lib/*:/opt/hadoop/share/hadoop/mapreduce/*:/opt/hadoop/contrib/capacity-scheduler/*.jar:/opt/hadoop/share/hadoop/tools/lib/*