用于 Google 云存储的 S3Guard 或 s3committer

Question

我在 Google Cloud Platform 上使用 Dataproc 和 Parquet，在 GCS 上处理数据，编写大量小到中等大小的文件非常麻烦，比使用较小的文件或 HDFS 慢几倍.

Hadoop 社区一直在开发 S3Guard，它使用 DynamoDB for S3A。同样，s3committer 使用 S3 的多部分 API 来提供一个更高效的简单替代提交器。

我正在寻找有关 GCS 的类似解决方案。 S3 的多部分 API 是 GCS 的 XML API 不提供的少数东西之一，因此不能按原样使用。相反，GCS 有一个“组合”API，您可以在其中单独上传文件，然后发出组合查询。这似乎可以用来调整来自s3committer 的分段上传，但我不太确定。

我找不到任何有关在 GCS 上使用 S3Guard 和备用键值存储（以及 S3A 连接器——甚至不确定它是否可以与 GCS XML API 一起使用）的信息。

0-rename 提交似乎是 Hadoop 和 Apache Spark 的常见问题。除了“写更少，更大的文件”之外，GCS 上的常见解决方案是什么？

Answer 1

这里有一些不同的东西在起作用。对于强制列表一致性的问题，Dataproc 传统上依赖于每个集群的 NFS 挂载来应用客户端强制写入后的列表一致性；最近，Google Cloud Storage 已设法改进其写后列表一致性语义，现在列表操作在所有写入后立即具有强一致性。 Dataproc 正在逐步淘汰客户端强制的一致性，GCS 不再需要像 DynamoDB 上的 S3Guard 这样的东西。

至于分段上传，理论上可以使用您提到的 GCS Compose，但在大多数情况下，单个大文件的并行分段上传在单流情况下最有帮助，而大多数 Hadoop/Spark 工作负载将已经在每台机器上并行化不同的任务，因此对每个单独的上传流进行多线程处理是没有好处的；无论是否使用并行分段上传，总吞吐量都将大致相同。

这样就留下了使用多部分 API 执行条件/原子提交的问题。 Hadoop 的 GCS 连接器目前确实使用了一种称为“可恢复上传”的东西，理论上一个节点可以负责“提交”一个由完全不同的节点上传的对象。客户端库目前的结构并未使其非常简单。但是，与此同时，GCS“重命名”的“复制和删除”阶段也与 S3 不同，因为它是作为元数据操作完成的，而不是真正的数据“复制”。这使得 GCS 可以使用 vanilla Hadoop FileCommitters，而不是需要“直接”提交到最终位置并跳过“_temporary”机制。必须“复制/删除”每个文件的元数据而不是真正的目录重命名可能并不理想，但它也与底层数据大小不成正比，仅与文件数量成正比。

当然，所有这些仍然不能解决提交大量小文件效率低下的事实。但是，它确实使“直接提交”方面可能不像您想象的那么重要。更常见的问题是 Hive 在完成时没有并行化文件提交，尤其是在提交到大量分区目录时。 Spark 在这方面要好得多，而且 Hive 应该会随着时间的推移而改进。

最近使用 native SSL library in Dataproc 1.2 进行了性能改进，您可以尝试使用开箱即用的 Dataproc 1.2，而无需“编写更少、更大的文件”。

否则，真正的解决方案确实涉及写入更少、更大的文件，因为即使您修复了写入端，如果您有太多小文件，读取端也会受到影响。 GCS 针对吞吐量进行了高度优化，因此任何小于 64MB 或 128MB 的东西都可能会花费更多的时间来处理启动任务和打开流与实际计算的开销（应该能够在 200ms-500ms 或所以）。

在这种情况下，您需要确保设置 hive.merge.mapfiles、hive.merge.mapredfiles 或 hive.merge.tezfiles（如果您正在使用它们），或者在保存到 GCS 之前重新分区您的 Spark 数据帧；合并到更大的分区通常是值得的，因为它可以使您的文件易于管理并从持续的更快读取中获益。

编辑：我忘记提及的一件事是，我一直在松散地使用术语 repartition，但在这种情况下，由于我们严格尝试将文件打包成更大的文件，你可能会使用 @ 做得更好改为 987654329@；另一个StackOverflow question about repartition vs coalese有更多讨论。

Answer 2

S3Guard，HADOOP-13345 通过让 DynamoDB 存储列表来改进 S3 的一致性。这使得第一次可以可靠地将 S3A 用作工作的直接目的地。否则，执行时间可能似乎是个问题，但真正的问题是基于重命名的提交者可能会得到不一致的列表，甚至看不到它必须重命名的文件。

S3Guard 提交者工作HADOOP-13786 将在完成后（截至 2017 年 8 月，仍在进行中）提供两个提交者。

暂存提交者

1. worker 写入本地文件系统
2. 任务提交者上传到 S3 但未完成操作。相反，它将提交元信息保存到 HDFS。
3. 此提交元信息作为 HDFS 中的普通任务/作业数据提交。
4. 在作业提交中，提交者从 HDFS 读取待提交的数据并完成它们，然后清理所有未完成的提交。

任务提交是所有数据的上传，时间为O(data/bandwidth)。

这是基于 Ryan 在 Netflix 的 s3committer，并且是最安全的一个。

魔术提交者

之所以被调用，是因为它在文件系统中发挥了“魔力”。

1. 文件系统本身识别像s3a://dest/__magic/job044/task001/__base/part-000.orc.snappy这样的路径
2. 将写入重定向到s3a://dest/__magic/job044/task001/__base/part-000.orc.snappy；未完成流 close() 调用中的写入。
3. 将提交元信息保存到 s3a，此处为 s3a://dest/__magic/job044/task001/__base/part-000.orc.snappy.pending

4. 任务提交：从该目录加载所有 .pending 文件，聚合，保存在别处。时间是O(files)；数据大小不重要。

5. 任务中止：加载所有 .pending 文件，中止提交

6. 作业提交：从已提交的任务中加载所有挂起的文件，完成。

因为它在 S3 中列出文件，所以它需要 S3Guard 来提供 AWS S3 上的一致性（其他 S3 实现是开箱即用的，所以不需要它）。

两个提交者共享相同的代码库，两者的作业提交将是O(files/threads)，因为它们都是不占用带宽或太多时间的短 POST 请求。

在测试中，暂存提交器比魔术提交器更快，因为魔术提交器与 S3 的对话更多，这很慢……尽管 S3Guard 加快了列表/getFileStatus 调用的速度。您写入的数据越多，暂存提交者的任务提交时间就越长，而对于相同数量的文件，魔术任务提交是恒定的。两者都比使用 rename() 更快，因为它是如何被列表模仿的，复制

GCS 和 Hadoop/Spark 提交算法

（这里的GCS代码我没有看，所以保留错误的权利。以Dennis Huo的说法为权威）

如果 GCS 确实比 S3A 复制然后删除更有效地重命名（），它应该更快，比 O（数据）更多的 O（文件），这取决于代码中的并行化。

我不知道他们是否可以选择 0-rename 提交者。 FileOutputFormat 下的 mapreduce 代码的变化是为了支持不同的/可插拔的提交者针对不同的文件系统，所以他们有机会在这里做点什么。

目前，请确保您使用的是 v2 MR 提交算法，该算法虽然对故障的恢复能力较差，但至少会将重命名推送到任务提交中，而不是作业提交中。

另见Spark and Object Stores。