您可以在 Spark/Hadoop 中将 s3:// 翻译（或别名）为 s3a:// 吗？

Posted 2023-04-15

【中文标题】您可以在 Spark/Hadoop 中将 s3:// 翻译（或别名）为 s3a:// 吗？

【英文标题】：Can you translate (or alias) s3:// to s3a:// in Spark/ Hadoop?

【发布时间】：2019-12-11 21:10:32

【问题描述】：

我们有一些代码在亚马逊服务器上运行，这些代码使用亚马逊建议的 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:// 路径，所以只需在 pyspark 中将它们转换为 s3a://。

我在 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])

【参考方案1】：

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

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

【讨论】：

你是说把 fs.s3.impl 设置为 org.apache.hadoop.fs.s3a.S3AFileSystem 就可以了？我稍后会试一试，但只是好奇你是否希望这能奏效。

我没有亲自尝试过，但我的理解是fs.X值决定了路径前缀。 impl 值只是用于fs.X 的任何接口

【参考方案2】：

cricket007 是正确的。

spark.hadoop.fs.s3.impl org.apache.fs.s3a.S3AFileSystem

org.apache.hadoop.FileSystem 中有一些代码从模式“s3”查找实现类，加载它并用完整的 URL 实例化它。

警告 核心 S3A FS 中没有特定代码用于查找 FS 模式为 s3a，但如果您使用 DynamoDB 一致性层“S3Guard”，您会遇到问题 - 这可能有点矫枉过正有人可以解决

【参考方案3】：

您可能无法配置 Spark 来帮助您“翻译”。

相反，这更像是一个设计问题。代码应该是可配置的，以便为不同的环境选择不同的协议（这就是我在类似情况下所做的）。如果你坚持在本地工作，一些代码重构可能无法避免……