python - 如何在以本地模式运行的 pyspark 中从 S3 读取数据？

Question

我正在使用 PyCharm 2018.1，它使用 Python 3.4，并在 virtualenv 中通过 pip 安装了 Spark 2.3。本地主机上没有安装hadoop，所以没有安装Spark(因此没有SPARK_HOME、HADOOP_HOME等)

当我尝试这样做时:

from pyspark import SparkConf
from pyspark import SparkContext
conf = SparkConf()\
    .setMaster("local")\
    .setAppName("pyspark-unittests")\
    .set("spark.sql.parquet.compression.codec", "snappy")
sc = SparkContext(conf = conf)
inputFile = sparkContext.textFile("s3://somebucket/file.csv")

我得到:

py4j.protocol.Py4JJavaError: An error occurred while calling o23.partitions.
: java.io.IOException: No FileSystem for scheme: s3

如果没有在本地完整安装 Hadoop，我如何在本地模式下运行 pyspark 时从 s3 读取数据？

FWIW - 当我以非本地模式在 EMR 节点上执行它时效果很好。

以下不起作用(同样的错误，尽管它确实解决并下载了依赖项):

import os
os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages "org.apache.hadoop:hadoop-aws:3.1.0" pyspark-shell'
from pyspark import SparkConf
from pyspark import SparkContext
conf = SparkConf()\
    .setMaster("local")\
    .setAppName("pyspark-unittests")\
    .set("spark.sql.parquet.compression.codec", "snappy")
sc = SparkContext(conf = conf)
inputFile = sparkContext.textFile("s3://somebucket/file.csv")

相同的(坏的)结果:

import os
os.environ['PYSPARK_SUBMIT_ARGS'] = '--jars "/path/to/hadoop-aws-3.1.0.jar" pyspark-shell'
from pyspark import SparkConf
from pyspark import SparkContext
conf = SparkConf()\
    .setMaster("local")\
    .setAppName("pyspark-unittests")\
    .set("spark.sql.parquet.compression.codec", "snappy")
sc = SparkContext(conf = conf)
inputFile = sparkContext.textFile("s3://somebucket/file.csv")

Answer 1

因此，Glennie 的回答很接近，但不是您的情况。关键是选择正确的依赖版本。如果你看看虚拟环境

一切都指向一个版本，2.7.3，您还需要使用它

os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages "org.apache.hadoop:hadoop-aws:2.7.3" pyspark-shell'

您应该通过检查项目虚拟环境中的路径 venv/Lib/site-packages/pyspark/jars 来验证安装使用的版本

然后您可以默认使用 s3a 或通过为其定义处理程序类来使用 s3

# Only needed if you use s3://
sc._jsc.hadoopConfiguration().set("fs.s3.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem")
sc._jsc.hadoopConfiguration().set('fs.s3a.access.key', 'awsKey')
sc._jsc.hadoopConfiguration().set('fs.s3a.secret.key', 'awsSecret')
s3File = sc.textFile("s3a://myrepo/test.csv")

print(s3File.count())
print(s3File.id())

下面是输出