local spark抱怨s3a模式

Question

我的问题有点similar to this one，但是建议的解决方案并没有解决我的问题。

我有一个非常简单的spark作业，我想在本地运行，基本上，它只是从S3读取一个文件并从中创建一个数据框。

当我在Amazon EC2集群上运行我的代码时，一切正常，但是当我想在本地运行它时，我得到这个错误

Caught exception while loading path, returning empty data frame: No FileSystem for scheme: s3a

以下是我在EC2集群和本地运行代码时的区别：

当我在本地运行它时，我注释了所有提供的spark依赖标记。

    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_2.11</artifactId>
        <version>2.1.0</version>
        <!--<scope>provided</scope>-->
    </dependency>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-sql_2.11</artifactId>
        <version>2.1.0</version>
        <!--<scope>provided</scope>-->
    </dependency>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-streaming_2.11</artifactId>
        <version>2.1.0</version>
        <!--<scope>provided</scope>-->
    </dependency>

另外，我还将这个添加到了依赖项中

<dependency>
  <groupId>org.apache.hadoop</groupId>
  <artifactId>hadoop-aws</artifactId>
  <version>3.0.0</version>
</dependency>

这就是我在本地和集群上创建spark会话的方式：

在集群上：

  def getSparkSession(config: BaseConfig): SparkSession = {

    val conf = new SparkConf()

    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    conf.set("spark.kryoserializer.buffer.mb","24")

    config.getSparkConfig.foreach(x => conf.set(x._1, x._2))

    val ss = SparkSession
      .builder()
      .config(conf)
      .getOrCreate()


    ss.sparkContext.hadoopConfiguration.set("fs.s3n.awsAccessKeyId", config.awsAccessKeyId)
    ss.sparkContext.hadoopConfiguration.set("fs.s3n.awsSecretAccessKey", config.awsSecretAccessKey)

    ss
  }

当我在本地运行时：

  def getLocalSparkSession(config: BaseConfig): SparkSession = {
    val warehouseLocation = new File("spark-warehouse").getAbsolutePath

    val ss = SparkSession.builder()
      .appName(this.getClass.getName)
      .master("local")
      .config("spark.sql.warehouse.dir", warehouseLocation)
      .config("spark.sql.shuffle.partitions", "4")
      .getOrCreate()
    ss.sparkContext.setLogLevel("WARN")


    ss.sparkContext.hadoopConfiguration.set("fs.s3n.awsAccessKeyId", config.awsAccessKeyId)
    ss.sparkContext.hadoopConfiguration.set("fs.s3n.awsSecretAccessKey", config.awsSecretAccessKey)

    ss
  }

附注:我的spark-shell --version显示它使用的是spark版本2.2.1，但我认为我在EC2集群上的spark是旧版本(应该是2.0左右)

Answer 1

不要将hadoop-aws版本与Spark构建时使用的版本混为一谈，它不起作用。您可能只看到了这方面的一个症状(在Hadoop2.7中，文件系统是自注册的，但在hadoop-2.8+中，它是在hadoop -common的core-default.xml文件中显式注册的。

理想情况下，您应该能够要求spark-hadoop-cloud模块随版本一起发布，但我在公共代码库中看不到这一点。

在此之前，确定hadoop版本，将该版本的hadoop-aws模块添加到您的构建中，并使用fs.s3a设置(包括密钥)。