使用 Dask 从 filesystem/S3 并行读取文件块？

Question

我正在整理一个概念证明，我想在其中使用 PyCuda 在分布式环境（具体来说是 AWS）中处理大型字符数据文件（每个任务一个文件约 8GB）。我知道 HDFS 会将数据文件分段并将其分发给工作人员，但我正在努力使我的环境尽可能简单，如果没有必要，我宁愿不必安装 Hadoop。

我最近看了几个来自 Continuum Analytics 的关于他们的 Dask 框架的网络研讨会，看起来它可以满足我的需要。鉴于以上段落和 Dask 框架，当前对文件系统的建议是什么？我是坚持使用 HDFS 还是有 better/simpler 解决方案？

Answer 1

大多数文件系统都提供只读取部分文件的能力，包括 HDFS、您的本地文件系统和 S3，AWS 实例的标准批量数据存储。这允许并行计算框架（如 Dask）将大文件分成许多较小的部分，供工作人员并行处理。

dask.bytes.read_bytes

对于大多数用例，这会在幕后自动发生（read_text 和 read_csv 的用户不必为此担心。）听起来您有自定义文件格式，所以我将引导您使用 read_bytes 函数。对于 S3，其工作方式如下：

from dask.bytes import read_bytes
sample, partitions = read_bytes('s3://bucket/keys.*.foo', 
                                blocksize=100000000)

Sample 将是一个 10kB 的简短数据样本，partitions 将是一个包含 dask.delayed 个对象的列表，您可以将这些对象与一般的 for 循环一起使用来构建您的计算。

如果您的数据有某种您希望 dask 遵守的分隔符，您可以使用 delimiter= 关键字参数提供它。

同样的功能也适用于其他系统，例如您的本地文件系统或 HDFS（如果您已经安装并导入了 hdfs3 和 distributed）。

sample, partitions = read_bytes('local://bucket/keys.*.foo', blocksize=100000000)
sample, partitions = read_bytes('hdfs://bucket/keys.*.foo')

例子

例如，这是我们如何实施的不正确但说明性的版本 dask.dataframe.read_csv

from dask import delayed
import pandas as pd
import dask.dataframe as dd

def read_csv(path, **kwargs):
    sample, partitions = read_bytes(path, blocksize=100000000, delimiter=b'\n')
    dataframes = [delayed(pd.read_csv)(part, **kwargs) for part in partitions]
    return dd.from_delayed(dataframes)

这是不正确的，因为 pd.read_csv 实际上需要一个 BytesIO 对象，我们没有稳健地处理关键字参数，并且我们没有很好地管理示例中的数据帧元数据（列、数据类型等）。这些细节妨碍了一般观点，可能超出了这个问题的兴趣。

编辑：在更常见的情况下使用其他函数

人们一直将此问题作为对 "How do I read data from S3?" 更一般性问题的回答大多数人不使用 read_bytes 界面，该界面有点低级。相反，大多数用户可能希望使用如下所示的高级功能之一：

import dask.bag as db
records = db.read_text('s3://bucket/keys.*.json').map(json.loads)

import dask.dataframe as dd
df = dd.read_csv('s3://bucket/keys.*.csv')

使用 Dask 从 filesystem/S3 并行读取文件块？

Read blocks of files in parallel from filesystem/S3 with Dask?

distributed-computing

dask

dask.bytes.read_bytes

例子

编辑：在更常见的情况下使用其他函数