3

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

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

4

1 回答 1

5

大多数文件系统都提供仅读取文件的一部分的能力，包括 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（如果您已经安装并导入了hdfs3and 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 对象，我们没有可靠地处理关键字参数，并且我们没有从示例中很好地管理数据帧元数据（列、dtypes 等）。这些细节妨碍了一般观点虽然并且可能超出了这个问题的兴趣。

编辑：在更常见的情况下使用其他功能

人们一直将此问题作为对“如何从 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')
于 2016-05-16T00:50:00.353 回答