从Spark 2.0访问S3

Question

我正在尝试从SparkSQL作业访问S3文件.我已经尝试了几个帖子的解决方案,但似乎没有任何效果.也许是因为我的EC2集群为Hadoop2.7运行了新的Spark2.0.

我用这种方式设置了hadoop:

sc.hadoopConfiguration.set("fs.s3a.impl","org.apache.hadoop.fs.s3a.S3AFileSystem")
sc.hadoopConfiguration.set("fs.s3a.awsAccessKeyId", accessKey)
sc.hadoopConfiguration.set("fs.s3a.awsSecretAccessKey", secretKey)

我使用sbt程序集构建一个超级jar:

name := "test"
version := "0.2.0"
scalaVersion := "2.11.8"

libraryDependencies += "com.amazonaws" % "aws-java-sdk" %   "1.7.4"
libraryDependencies += "org.apache.hadoop" % "hadoop-aws" % "2.7.3" excludeAll(
    ExclusionRule("com.amazonaws", "aws-java-sdk"),
    ExclusionRule("commons-beanutils")
)

libraryDependencies += "org.apache.spark" %% "spark-core" % "2.0.0" % "provided"
libraryDependencies += "org.apache.spark" %% "spark-sql" % "2.0.0" % "provided"

当我将作业提交到群集时,我总是遇到以下错误:

线程"main"中的异常org.apache.spark.SparkException:由于阶段失败而中止作业:阶段0.0中的任务0失败4次,最近失败:阶段0.0中丢失的任务0.3(TID 6,172.31.7.246):java .lang.RuntimeException:java.lang.ClassNotFoundException:在org.apache.hadoop的org.apache.hadoop.conf.Configuration.getClass(Configuration.java:2195)中找不到类org.apache.hadoop.fs.s3a.S3AFileSystem .fs.FileSystem.getFileSystemClass(FileSystem.java:2638)org.apache.hadoop.fs.FileSystem.createFileSystem(FileSystem.java:2651)org.apache.hadoop.fs.FileSystem.access $ 200(FileSystem.java: 92)org.apache.hadoop.fs.FileSystem $ Cache.getInternal(FileSystem.java:2687)atg.apache.hadoop.fs.FileSystem $ Cache.get(FileSystem.java:2669)org.apache.hadoop org.apache.spark.util.Utils $ .getHadoopFileSystem(Utils.scala:1726)中的.fs.FileSystem.get(FileSystem.java:371)org.apache.spark.util.Utils $ .doFetchFile(Utils.scala) :662)at org.apache.spark.util.Utils $ .fetchFile(Utils.scala:446)在org.apache.spark.executor.Executor $$ anonfun $ org $ apache $ spark $ executor $ Executor $$ updateDependencies $ 3.apply(Executor.scala:476)

看起来驱动程序能够从S3中读取而没有问题,但不能读取工作人员/执行者......我不明白为什么我的uberjar不够用.

但是,我尝试使用以下命令配置spark-submit,但没有成功:

--packages com.amazonaws:aws-java-sdk:1.7.4,org.apache.hadoop:hadoop-aws:2.7.3

PS:如果我切换到s3n协议,我得到以下异常:

java.io.IOException:没有用于scheme的文件系统:s3n

Answer 1

如果你想使用s3n:

sc.hadoopConfiguration.set("fs.s3n.impl","org.apache.hadoop.fs.s3native.NativeS3FileSystem")
sc.hadoopConfiguration.set("fs.s3n.awsAccessKeyId", accessKey)
sc.hadoopConfiguration.set("fs.s3n.awsSecretAccessKey", secretKey)

现在,关于异常,您需要确保两个JAR都在驱动程序和工作线类路径上,并且如果您通过--jars标志使用客户端模式,请确保将它们分发到工作节点:

spark-submit \
--conf "spark.driver.extraClassPath=/location/to/aws-java-sdk.jar" \
--conf "spark.driver.extraClassPath=/location/to/hadoop-aws.jar" \
--jars /location/to/aws-java-sdk.jar,/location/to/hadoop-aws.jar \

此外,如果您正在构建您的超级JAR并且包含aws-java-sdk并且hadoop-aws没有理由使用该--packages标志.

Answer 2

实际上spark的所有操作都在worker上进行。并且您在 master 上设置了这些配置，因此一旦您可以尝试在 mapPartition{ } 上应用 s3 的配置