标签: dask

我有27GB的2D Tiff文件，它们代表3D图像电影的片段。我希望能够像分割一个简单的numpy4d数组一样对这些数据进行切片。看起来dask.array是一个很好的工具，用于将数组作为hdf5文件存储在内存中后即可对其进行干净处理。

如果这些文件不能全部放入内存，我该如何首先将它们存储为hdf5文件。我是h5.py和数据库的新手。

谢谢。

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

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

我正在尝试使用get_dummiesviadask但它不会转换我的变量，也不会出错：

>>> import dask.dataframe as dd
>>> import pandas as pd
>>> df_d = dd.read_csv('/datasets/dask_example/dask_get_dummies_example.csv')
>>> df_d.head()
   uid gender
0    1      M
1    2    NaN
2    3    NaN
3    4      F
4    5    NaN
>>> daskDataCategorical = df_d[['gender']]
>>> daskDataDummies = dd.get_dummies(daskDataCategorical) 
>>> daskDataDummies.head()
  gender
0      M
1    NaN
2    NaN
3      F
4    NaN
>>> daskDataDummies.compute() 
  gender
0      M
1    NaN
2    NaN
3      F
4    NaN
5      F
6      M
7      F
8      M …

如何以“有效”的方式在 Dask 中执行以下操作：

我的 dask 数据框有“日期”（日期时间）、“MAC”（类别）和“ID”（整数）列，它们已经按日期排序，我想获得一个新列，其中包含 http 请求到达的时间增量对于给定的 mac 地址。

在 Pandas 中，我会执行以下操作：设置一个多索引 ['ID', 'MAC']，然后使用df['Date'].diff(1). 似乎不支持多索引，因此无法重现相同的解决方案。重要的部分是必须在给定请求和以下记录器（按日期排序）之间对同一 MAC 进行差异。

我想知道我是否可以使用 dask 代替 Pandas。我可能面临哪些问题？

1）我猜对于较小的数据集，dask 会比 Pandas 慢。我对此没有意见，因为有时我不知道数据的大小，也不知道服务器配置。

2）我将不得不学习稍微不同的语法（例如计算）

我会遇到 dask 数据框不能做熊猫数据框可以做的事情的情况吗？

我试图将一个大熊猫数据帧作为函数参数传递给分布式 dask 的工作人员。我尝试过的（X 是我的数据框）：

1 将数据直接传递给函数：

def test(X):
    return X
f=client.submit(test, X)
f.result()

2 在初始化函数中保存数据帧。

def worker_init(r_X):
    global X
    X=r_X
client.run(worker_init,X,y)

3 将数据帧分散到所有节点，然后通过期货使用它

def test(X):
    return X
f_X = client.scatter(X, broadcast=True)
f = client.submit(test,f_X)
f.result()

没有一个变体适用于我的情况。变体 1 和 2 的工作方式几乎相同。dask-scheduler 为每个任务增加内存，并且永远不会释放它，直到它耗尽内存并且任务失败。

变体 3 不起作用，因为我没有传递 Pandas 数据帧，而是得到了一些垃圾。

如何将数据帧发送给工作人员并且在调度程序上没有 MemoryError？

变体 3 的完整代码应该是内存高效的，但甚至不传递数据帧：

import pandas as pd
import numpy as np
from distributed import Client
client = Client('localhost:8786')
X = np.random.rand(10000,100)
X=pd.DataFrame(X)
f_X = client.scatter(X, broadcast=True)
def test(X):
    return X
f = …

pd.DatetimeIndex(df_dask_dataframe['name_col'])

我有一个 dask 数据框，我想将带有日期的列转换为日期时间索引。但是我得到一个未实现的错误。有解决方法吗？

我试图找到一种开始在 Pandas 中处理非常大的 CSV 文件的方法，最终能够使用 XGBoost 进行一些机器学习。

我在使用 mySQL 或一些 sqllite 框架来管理我的数据块之间挣扎；我的问题是稍后的机器学习方面，以及一次加载块以训练模型。

我的另一个想法是使用Dask，它是由 Pandas 构建的，但也具有 XGBoost 功能。

我不确定最好的起点是什么，并希望征求意见！我倾向于，Dask但我还没有使用它。

我有一个Dask数据框，其中一列包含一个浮点数的numpy数组：

import dask.dataframe as dd
import pandas as pd
import numpy as np

df = dd.from_pandas(
    pd.DataFrame(
        {
            'id':range(1, 6),
            'vec':[np.array([1.0, 2.0, 3.0, 4.0, 5.0])] * 5
        }), npartitions=1)

df.compute()

   id                        vec
0   1  [1.0, 2.0, 3.0, 4.0, 5.0]
1   2  [1.0, 2.0, 3.0, 4.0, 5.0]
2   3  [1.0, 2.0, 3.0, 4.0, 5.0]
3   4  [1.0, 2.0, 3.0, 4.0, 5.0]
4   5  [1.0, 2.0, 3.0, 4.0, 5.0]

如果我尝试将其写为实木复合地板，则会出现错误：

df.to_parquet('somefile')
....
Error converting column "vec" to bytes using encoding UTF8. Original …

更新：

pandas df是这样创建的：

df = pd.read_sql(query, engine)
encoded = pd.get_dummies(df, columns=['account'])

从此df创建一个dask df如下所示：

df = dd.from_pandas(encoded, 50)

轻而易举地执行操作不会导致可见的进度（使用轻便的诊断程序进行检查）：

result = df.groupby('journal_entry').max().reset_index().compute()

原版的：

我有一个具有270万行和4,000列的大熊猫df。除四列外，所有列均为dtype uint8。uint8列仅保留值1或0。我试图在df上执行此操作：

result = df.groupby('id').max().reset_index()

可以预期，此操作将立即返回内存错误。我最初的想法是在水平和垂直方向上对df进行分块。但是，这会造成混乱，因为.max()需要跨所有uint8列（而不仅仅是一对列）执行该操作。此外，像这样对df进行分块仍然非常慢。我的机器上有32 GB的RAM。

哪种策略可以减轻内存问题？