multithreading - Spark/EMR 可以从 s3 多线程读取数据吗

Question

由于一些不幸的事件序列，我们最终在 s3 上存储了一个非常分散的数据集。表元数据存储在Glue上，数据用“bucketBy”写入，以parquet格式存储。因此文件的发现不是问题，并且 Spark 分区的数量等于桶的数量，这提供了良好的并行度。

当我们在 Spark/EMR 上加载这个数据集时，我们最终让每个 spark 分区从 s3 加载大约 8k 个文件。

由于我们以列格式存储数据；根据我们需要几个字段的用例，我们并没有真正读取所有数据，而是读取存储内容的很小一部分。

根据工作节点上的 CPU 利用率，我可以看到每个任务(每个分区运行)都使用了大约 20% 的 CPU，我怀疑这是由于每个任务的单个线程从 s3 顺序读取文件，所以很多等等...

有没有办法鼓励 EMR 上的 spark 任务从 s3 多线程读取数据，以便我们可以在一个任务中同时从 s3 读取多个文件？这样，我们可以利用 80% 空闲的 CPU 来让事情变得更快一点吗？

Answer 1

使用 Spark 数据帧读取 S3 数据有两个部分:

发现(列出 S3 上的对象)

读取S3对象，包括解压等

发现通常发生在驱动程序上。一些托管 Spark 环境具有使用集群资源进行更快发现的优化。除非超过 100K 对象，否则这通常不是问题。如果您有 .option("mergeSchema", true)，发现会更慢因为每个文件都必须接触才能发现其架构。

读取 S3 文件是执行操作的一部分。读取的并行度为 min(分区数，可用内核数)。更多的分区 + 更多的可用内核意味着更快的 I/O……理论上。实际上，如果您没有为 S3 定期访问这些文件以扩展其可用性，则 S3 可能会非常慢。因此，在实践中，额外的 Spark 并行性 yield 递减。观察每个事件核心的总网络 RW 带宽并调整您的执行以获得最高值(value)。

您可以通过 df.rdd.partitions.length 发现分区数.

如果 S3 I/O 吞吐量较低，您还可以执行以下操作:

确保 S3 上的数据在涉及其前缀时是分散的(请参阅 https://docs.aws.amazon.com/AmazonS3/latest/dev/optimizing-performance.html )。

打开 AWS 支持请求并询问要扩展的数据的前缀。

试验不同的节点类型。我们发现存储优化的节点具有更有效的 I/O。

希望这可以帮助。