apache-spark - 你能翻译(或别名)s3 ://to s3a://in Spark/Hadoop?

Question

我们有一些代码在亚马逊服务器上运行，这些代码使用亚马逊建议的 s3://方案加载 Parquet 。但是，一些开发人员希望在 Windows 上使用 spark 安装在本地运行代码，但 spark 固执地坚持使用 s3a://方案。

我们可以使用 s3a 很好地读取文件，但是我们得到一个 java.lang.NoClassDefFoundError: org/jets3t/service/S3ServiceException。

SparkSession available as 'spark'.
>>> spark.read.parquet('s3a://bucket/key')
DataFrame[********************************************]
>>> spark.read.parquet('s3://bucket/key')
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "C:\spark\spark-2.4.4-bin-hadoop2.7\python\pyspark\sql\readwriter.py", line 316, in parquet
    return self._df(self._jreader.parquet(_to_seq(self._spark._sc, paths)))
  File "C:\spark\spark-2.4.4-bin-hadoop2.7\python\lib\py4j-0.10.7-src.zip\py4j\java_gateway.py", line 1257, in __call__
  File "C:\spark\spark-2.4.4-bin-hadoop2.7\python\pyspark\sql\utils.py", line 63, in deco
    return f(*a, **kw)
  File "C:\spark\spark-2.4.4-bin-hadoop2.7\python\lib\py4j-0.10.7-src.zip\py4j\protocol.py", line 328, in get_return_value
py4j.protocol.Py4JJavaError: An error occurred while calling o37.parquet.
: java.lang.NoClassDefFoundError: org/jets3t/service/S3ServiceException
        at org.apache.hadoop.fs.s3.S3FileSystem.createDefaultStore(S3FileSystem.java:99)
        at org.apache.hadoop.fs.s3.S3FileSystem.initialize(S3FileSystem.java:89)
        at org.apache.hadoop.fs.FileSystem.createFileSystem(FileSystem.java:2669)
        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)
        at org.apache.spark.sql.execution.streaming.FileStreamSink$.hasMetadata(FileStreamSink.scala:45)
        at org.apache.spark.sql.execution.datasources.DataSource.resolveRelation(DataSource.scala:332)
        at org.apache.spark.sql.DataFrameReader.loadV1Source(DataFrameReader.scala:223)
        at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:211)
        at org.apache.spark.sql.DataFrameReader.parquet(DataFrameReader.scala:644)
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(Unknown Source)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source)
        at java.lang.reflect.Method.invoke(Unknown Source)
        at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
        at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
        at py4j.Gateway.invoke(Gateway.java:282)
        at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
        at py4j.commands.CallCommand.execute(CallCommand.java:79)
        at py4j.GatewayConnection.run(GatewayConnection.java:238)
        at java.lang.Thread.run(Unknown Source)
Caused by: java.lang.ClassNotFoundException: org.jets3t.service.S3ServiceException
        at java.net.URLClassLoader.findClass(Unknown Source)
        at java.lang.ClassLoader.loadClass(Unknown Source)
        at sun.misc.Launcher$AppClassLoader.loadClass(Unknown Source)
        at java.lang.ClassLoader.loadClass(Unknown Source)
        ... 24 more

有没有办法让hadoop或spark或pyspark通过某种神奇的配置将URI方案从s3“翻译”到s3a？更改代码不是我们喜欢的选择，因为它会涉及大量测试。

本地环境是 windows 10、pyspark2.4.4，安装了 hadoop2.7(预构建)、python3.7.5 和正确的 aws 库。

编辑:我使用的一个技巧 - 因为我们不应该使用 s3://路径，所以只需将它们转换为 s3a://在 pyspark 中。

我在 readwriter.py 中添加了以下函数，并在任何有路径调用 jvm 的地方调用它。工作正常，但如果这是一个配置选项会很好。

def massage_paths(paths):
    if isinstance(paths, basestring):
        return 's3a' + x[2:] if x.startswith('s3:') else x
    if isinstance(paths, list):
        t = list
    else:
        t = tuple
    return t(['s3a' + x[2:] if x.startswith('s3:') else x for x in paths])

Answer 1

理想情况下，您可以重构代码以检测运行时环境，或将路径外部化为可在各个区域使用的配置文件。

否则，您需要编辑 hdfs-site.xml 来配置 fs.s3a.impl将 s3a 重命名为 s3 的键，您也许可以保持该值相同。所有 Spark 工作人员都需要进行这种更改