从Apache Spark访问公共可用的Amazon S3文件

Question

我有一个公共可用的Amazon s3资源（文本文件）并希望从spark中访问它。这意味着 - 我没有任何凭据亚马逊 - 它工作得很好，如果我想只要下载​​：

val bucket = "<my-bucket>" 
val key = "<my-key>" 

val client = new AmazonS3Client 
val o = client.getObject(bucket, key) 
val content = o.getObjectContent // <= can be read and used as input stream 

然而，当我尝试从火花背景

val conf = new SparkConf().setAppName("app").setMaster("local") 
val sc = new SparkContext(conf) 
val f = sc.textFile(s"s3a://$bucket/$key") 
println(f.count()) 

访问同一资源我收到堆栈跟踪以下错误：

Exception in thread "main" com.amazonaws.AmazonClientException: Unable to load AWS credentials from any provider in the chain 
    at com.amazonaws.auth.AWSCredentialsProviderChain.getCredentials(AWSCredentialsProviderChain.java:117) 
    at com.amazonaws.services.s3.AmazonS3Client.invoke(AmazonS3Client.java:3521) 
    at com.amazonaws.services.s3.AmazonS3Client.headBucket(AmazonS3Client.java:1031) 
    at com.amazonaws.services.s3.AmazonS3Client.doesBucketExist(AmazonS3Client.java:994) 
    at org.apache.hadoop.fs.s3a.S3AFileSystem.initialize(S3AFileSystem.java:297) 
    at org.apache.hadoop.fs.FileSystem.createFileSystem(FileSystem.java:2653) 
    at org.apache.hadoop.fs.FileSystem.access$200(FileSystem.java:92) 
    at org.apache.hadoop.fs.FileSystem$Cache.getInternal(FileSystem.java:2687) 
    at org.apache.hadoop.fs.FileSystem$Cache.get(FileSystem.java:2669) 
    at org.apache.hadoop.fs.FileSystem.get(FileSystem.java:371) 
    at org.apache.hadoop.fs.Path.getFileSystem(Path.java:295) 
    at org.apache.hadoop.mapred.FileInputFormat.listStatus(FileInputFormat.java:221) 
    at org.apache.hadoop.mapred.FileInputFormat.getSplits(FileInputFormat.java:270) 
    at org.apache.spark.rdd.HadoopRDD.getPartitions(HadoopRDD.scala:207) 
    at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:219) 
    at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:217) 
    at scala.Option.getOrElse(Option.scala:121) 
    at org.apache.spark.rdd.RDD.partitions(RDD.scala:217) 
    at org.apache.spark.rdd.MapPartitionsRDD.getPartitions(MapPartitionsRDD.scala:32) 
    at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:219) 
    at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:217) 
    at scala.Option.getOrElse(Option.scala:121) 
    at org.apache.spark.rdd.RDD.partitions(RDD.scala:217) 
    at org.apache.spark.SparkContext.runJob(SparkContext.scala:1781) 
    at org.apache.spark.rdd.RDD.count(RDD.scala:1099) 
    at com.example.Main$.main(Main.scala:14) 
    at com.example.Main.main(Main.scala) 
    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:497) 
    at com.intellij.rt.execution.application.AppMain.main(AppMain.java:140) 

我不想提供任何AWS凭据 - 我只是想访问资源匿名（现在） - 如何实现这一目标？我可能需要使它像AnonymousAWSCredentialsProvider - 但如何把它放在火花或hadoop？

P.S.我的build.sbt以防万一

scalaVersion := "2.11.7" 

libraryDependencies ++= Seq(
    "org.apache.spark" %% "spark-core" % "1.4.1", 
    "org.apache.hadoop" % "hadoop-aws" % "2.7.1" 
) 

更新：我做了一些调查后 - 我看到了它为什么没有工作的原因。

首先，S3AFileSystem创建AWS客户端凭证的顺序如下：

AWSCredentialsProviderChain credentials = new AWSCredentialsProviderChain(
    new BasicAWSCredentialsProvider(accessKey, secretKey), 
    new InstanceProfileCredentialsProvider(), 
    new AnonymousAWSCredentialsProvider() 
); 

“ACCESSKEY”和“秘密密钥”值从火花的conf例如采取（密钥必须“fs.s3a。 access.key“和”fs.s3a.secret.key“org.apache.hadoop.fs.s3a.Constants.ACCESS_KEY和org.apache.hadoop.fs.s3a.Constants.SECRET_KEY常量，这样更方便）。

其次 - 您可能会看到AnonymousAWSCredentialsProvider是第三个选项（最后优先级） - 可能会出现什么错误？请参见AnonymousAWSCredentials的实现：

public class AnonymousAWSCredentials implements AWSCredentials { 

    public String getAWSAccessKeyId() { 
     return null; 
    } 

    public String getAWSSecretKey() { 
     return null; 
    } 
} 

它对访问密钥和密钥都只是返回null。听起来很合理。但是，看看里面AWSCredentialsProviderChain：

AWSCredentials credentials = provider.getCredentials(); 

if (credentials.getAWSAccessKeyId() != null && 
    credentials.getAWSSecretKey() != null) { 
    log.debug("Loading credentials from " + provider.toString()); 

    lastUsedProvider = provider; 
    return credentials; 
} 

它不选择供应商的情况下，两个键是空 - 这意味着匿名凭据不能工作。看起来像aws-java-sdk-1.7.4中的一个bug。我试图使用最新版本 - 但它与hadoop-aws-2.7.1不兼容。

还有其他想法吗？

Answer 1

我个人从未访问过Spark的公共数据。您可以尝试使用虚拟凭证，或仅为此用途创建虚拟凭证。直接在SparkConf对象上设置它们。

val sparkConf: SparkConf = ??? 
val accessKeyId: String = ??? 
val secretAccessKey: String = ??? 
sparkConf.set("spark.hadoop.fs.s3.awsAccessKeyId", accessKeyId) 
sparkConf.set("spark.hadoop.fs.s3n.awsAccessKeyId", accessKeyId) 
sparkConf.set("spark.hadoop.fs.s3.awsSecretAccessKey", secretAccessKey) 
sparkConf.set("spark.hadoop.fs.s3n.awsSecretAccessKey", secretAccessKey) 

作为替代，读取DefaultAWSCredentialsProviderChain文件，看看那里的凭据寻找。该列表（顺序很重要）是：

• 环境变量 - AWS_ACCESS_KEY_ID和AWS_SECRET_KEY
• Java系统属性 - aws.accessKeyId和aws.secretKey
• 凭据配置文件的文件在默认位置（〜/ 。AWS /凭证）的所有AWS的SDK和AWS CLI
• 实例档凭证，通过亚马逊EC2元数据服务