相关疑难解决方法(0)

我正在尝试从Spark AWS shell中加载来自Amazon AWS S3存储桶的数据.

我咨询过以下资源:

使用Apache Spark从Amazon S3解析文件

如何从Apache Spark访问s3a://文件？

Hortonworks Spark 1.6和S3

Cloudera的

自定义s3端点

我已经下载并解压缩了Apache Spark 2.2.0.在conf/spark-defaults我有以下(注意我替换access-key和secret-key):

spark.hadoop.fs.s3a.impl=org.apache.hadoop.fs.s3a.S3AFileSystem
spark.hadoop.fs.s3a.access.key=access-key 
spark.hadoop.fs.s3a.secret.key=secret-key

我已经下载hadoop-aws-2.8.1.jar并aws-java-sdk-1.11.179.jar从mvnrepository,并将它们放置在jars/目录中.然后我启动Spark shell:

bin/spark-shell --jars jars/hadoop-aws-2.8.1.jar,jars/aws-java-sdk-1.11.179.jar

在shell中,以下是我尝试从S3存储桶加载数据的方法:

val p = spark.read.textFile("s3a://sparkcookbook/person")

以下是导致的错误:

java.lang.NoClassDefFoundError: org/apache/hadoop/fs/GlobalStorageStatistics$StorageStatisticsProvider
  at java.lang.Class.forName0(Native Method)
  at java.lang.Class.forName(Class.java:348)
  at org.apache.hadoop.conf.Configuration.getClassByNameOrNull(Configuration.java:2134)
  at org.apache.hadoop.conf.Configuration.getClassByName(Configuration.java:2099)
  at org.apache.hadoop.conf.Configuration.getClass(Configuration.java:2193)
  at org.apache.hadoop.fs.FileSystem.getFileSystemClass(FileSystem.java:2654)
  at org.apache.hadoop.fs.FileSystem.createFileSystem(FileSystem.java:2667)
  at org.apache.hadoop.fs.FileSystem.access$200(FileSystem.java:94)
  at org.apache.hadoop.fs.FileSystem$Cache.getInternal(FileSystem.java:2703)
  at org.apache.hadoop.fs.FileSystem$Cache.get(FileSystem.java:2685)
  at org.apache.hadoop.fs.FileSystem.get(FileSystem.java:373)
  at org.apache.hadoop.fs.Path.getFileSystem(Path.java:295)

当我尝试按如下方式启动Spark shell时:

bin/spark-shell --packages …

我整个上午都在寻找这个问题的明确答案，但找不到任何可以理解的东西。我刚开始使用 pyspark（用 pip 安装），有一个简单的 .py 文件从本地存储读取数据，在本地做一些处理和写入结果。我目前正在使用：python my_file.py

我想要做的是： 使用 AWS S3 中的文件作为输入，将结果写入 AWS3 上的存储桶

我能够使用“boto3”创建一个存储桶和加载文件，但看到一些使用“spark.read.csv”的选项，我想使用它。

我尝试过的： 我尝试使用以下方法设置凭据：

spark = SparkSession.builder \
            .appName("my_app") \
            .config('spark.sql.codegen.wholeStage', False) \
            .getOrCreate()\

spark._jsc.hadoopConfiguration().set("fs.s3a.awsAccessKeyId", "my_key_id")
spark._jsc.hadoopConfiguration().set("fs.s3a.awsSecretAccessKey", "my_secret_key")

然后 ：

df = spark.read.option("delimiter", ",").csv("s3a://bucket/key/filename.csv", header = True)

但得到错误：

java.io.IOException: No FileSystem for scheme: s3a

问题 ：

• 我是否需要特别安装一些东西才能启用 pyspark S3？
• 我应该以某种方式打包我的代码并使用 pyspark 控制台运行一个特殊的命令吗？

谢谢大家，抱歉重复的问题

解决了 ：

解决方案如下：

要将本地 spark 实例链接到 S3，您必须将 aws-sdk 和 hadoop-sdk 的 jar 文件添加到您的类路径并使用以下命令运行您的应用程序：spark-submit --jars my_jars.jar

请注意您用于 SDK 的版本，并非所有版本都兼容：aws-java-sdk-1.7.4、hadoop-aws-2.7.4 对我有用。

我使用的配置是：

spark = SparkSession.builder \ …

我想知道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

谢谢!

我将大量数据上传到我想要的 S3 存储桶，以便使用 Spark 和 Zeppelin 进行分析/可视化。然而，我仍然坚持从 S3 加载数据。

我做了一些阅读，以便将它们整合在一起，并免除我血腥的细节。我使用docker容器p7hb/docker-spark作为 Spark 安装，我从 S3 读取数据的基本测试来自这里：

1. 我启动容器以及其中的主进程和从进程。我可以通过查看暴露在端口 8080 上的 Spark Master WebUI 来验证这项工作。此页面确实列出了工作人员并在标题“已完成的应用程序”下保留了我所有失败尝试的日志。所有这些都在状态中FINISHED。

2. 我bash在该容器内打开一个并执行以下操作：

   a)按照此处的建议导出环境变量AWS_ACCESS_KEY_ID和。AWS_SECRET_ACCESS_KEY

   b) 开始spark-shell。为了访问 S3，似乎需要加载一些额外的包。在浏览 SE 时，我特别发现了这个，它教会了我，我可以使用该--packages参数来加载所述包。基本上我运行spark-shell --packages com.amazonaws:aws-java-sdk:1.7.15,org.apache.hadoop:hadoop-aws:2.7.5(, 对于版本的任意组合)。

   c) 我运行以下代码

   sc.hadoopConfiguration.set("fs.s3a.endpoint", "s3-eu-central-1.amazonaws.com") sc.hadoopConfiguration.set("fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem") sc.hadoopConfiguration.set("com.amazonaws.services.s3.enableV4", "true")

   val sonnets=sc.textFile("s3a://my-bucket/my.file")

   val counts = sonnets.flatMap(line => line.split(" ")).map(word => (word, 1)).reduceByKey(_ + _)

然后我收到各种不同的错误消息，具体取决于我在 2b) 中选择的版本。

我想 …

有一个用例，我们想要从具有 JSON 的 S3 读取文件。然后，根据特定的 JSON 节点值，我们希望对数据进行分组并将其写入 S3。

我能够读取数据，但无法找到关于如何基于 JSON 密钥对数据进行分区然后上传到 S3 的好示例。任何人都可以提供任何示例或向我指出可以帮助我解决此用例的教程吗？

创建数据框后，我得到了数据的架构：

root
 |-- customer: struct (nullable = true)
 |    |-- customerId: string (nullable = true)
 |-- experiment: string (nullable = true)
 |-- expiryTime: long (nullable = true)
 |-- partitionKey: string (nullable = true)
 |-- programId: string (nullable = true)
 |-- score: double (nullable = true)
 |-- startTime: long (nullable = true)
 |-- targetSets: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- …

我正在做一些关于spark_with_hadoop2.7（2.4.3），hadoop（3.2.0）和Ceph luminous的组合的事情。当我尝试使用spark访问ceph时（例如，spark-sql在shell上启动），出现如下异常：

 INFO impl.MetricsSystemImpl: s3a-file-system metrics system started
Exception in thread "main" java.lang.NoSuchMethodError: org.apache.hadoop.security.ProviderUtils.excludeIncompatibleCredentialProviders(Lorg/apache/hadoop/conf/Configuration;Ljava/lang/Class;)Lorg/apache/hadoop/conf/Configuration;
        at org.apache.hadoop.fs.s3a.S3AUtils.getAWSAccessKeys(S3AUtils.java:740)
        at org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider.<init>(SimpleAWSCredentialsProvider.java:58)
        at org.apache.hadoop.fs.s3a.S3AUtils.createAWSCredentialProviderSet(S3AUtils.java:600)

对于NoSuchMethodError，这很可能是因为根据how-do-i-fix-a-nosuchmethoderror编译的类版本与运行的类版本不同。

要访问， 实际上使用了Cephaws相关的jaraws-java-sdk-bundle-1.11.375.jar及其hadoop-aws-3.2.0.jar下的文件。$HADOOP_HOME/share/hadoop/tools/lib我做了以下操作：

1、将这两个jar复制到$SPARK_HOME/jars
2、修改$HADOOP_HOME/etc/hadoop/hadoop-env.sh添加以下语句：

export HADOOP_CLASSPATH=$HADOOP_CLASSPATH:$HADOOP_HOME/share/hadoop/tools/lib/*

通过执行上述步骤，我可以启动 hdfs 来访问 ceph，例如，我可以用来hdfs dfs -ls列出 ceph 存储桶下的文件夹。它证明与aws相关的jar工作正常。（正如我的理解）。

但是为什么当我调用 Spark 时会抛出有关 aws s3a 的异常？