标签: dask

对于当前的项目，我计划将两个非常大的 CSV 文件与 Dask 合并，作为 Pandas 的替代方案。我已经彻底安装了Dask pip install "dask[dataframe]"。

然而，在跑步时import dask.dataframe as dd，我收到了反馈ModuleNotFoundError: No module named 'dask.dataframe'; 'dask' is not a package。

几个用户似乎遇到了同样的问题，并建议通过 Conda 安装该模块，这对我的情况也没有帮助。

找不到模块是什么原因？

我有一个使用 Dask 编写的大型数据提取作业，其中每个任务将从数十个数据库的大量表中查询一个表。对于每个数据库实例，我想限制一次连接的任务数量（即限制）。例如，我可能有 100 个任务连接到数据库 A，100 个任务连接到数据库 B，100 个任务连接到数据库 C，等等，并且我希望确保在任何给定时间连接到任何数据库的任务不超过 20 个。

我发现 Dask 提供了基于工作线程资源（CPU、MEM、GPU 等）的约束，但是数据库资源是“全局”的，因此对于任何 Dask 工作线程来说都不是特定的。Dask 是否提供任何方法来对任务并发的此类约束进行建模？

我对达斯克有疑问。我已经检查了 csv 文件，一切正常，我不上传它，因为它是机密的。但也许您可以尝试自己的 CSV 并看到您收到相同的错误。

我的代码如下：

from dask.distributed import Client
client = Client(n_workers=4)
client

import dask.dataframe as dd
df = dd.read_csv('merged_data.csv')

X=df[['Mp10','Mp10_cal','Mp2_5','Mp2_5_cal','Humedad','Temperatura']]

y = df['Sector']

from dask_ml.model_selection import train_test_split
X_train, X_test, y_train, y_test = train_test_split(X, y, train_size=0.8, random_state=42, shuffle=False)

import joblib
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import GridSearchCV

from sklearn.model_selection import GridSearchCV
# Create the parameter grid based on the results of random search 
param_grid = {
    'bootstrap': [True],
    'max_depth': [80, 90, 100, 110],
    'max_features': [2, 3],
    'min_samples_leaf': [3, …

我最近发现了Dask。我对Dask Dataframe和其他数据结构有非常基本的问题。

1. Dask Dataframe是不可变的数据类型吗？
2. Dask数组和Dataframe是惰性数据结构吗？

我不知道是针对我的情况使用dask还是spark或pandas。我有200 GB的数据要计算。使用普通的python程序花费了9个小时来计算操作。但是通过使用16核处理器，它可以在较短的时间内并行处理。如果将数据框划分为大熊猫，则需要担心计算的可交换性和关联性。另一方面，我可以使用独立的Spark集群来拆分数据并并行运行。

我需要像在Spark中一样在Dask中设置任何集群吗？
如何在我自己的计算节点中运行Dask数据帧？
Dask是否需要主从设置？

我是大熊猫的粉丝，所以我正在寻找与大熊猫类似的解决方案。

我尝试导入时遇到错误,dask.dot无法找到graphviz安装.但是,安装了graphviz和pygraphviz.

balter@exalab3:~$ conda install dask
Fetching package metadata ...........
Solving package specifications: ..........

# All requested packages already installed.
# packages in environment at /home/.../miniconda3:
#
dask                      0.13.0                   py35_0    conda-forge
balter@exalab3:~$ python
Python 3.5.3 | packaged by conda-forge | (default, Jan 23 2017, 19:01:48)
[GCC 4.8.2 20140120 (Red Hat 4.8.2-15)] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> import dask.dot
Traceback (most recent call last):
  File "/home/.../miniconda3/lib/python3.5/site-packages/dask/utils.py", line 63, in import_required
    return import_module(mod_name)
  File …

我发现dask.dataframe.rename无论何时从镶木地板文件中读取数据帧,都会显着增加计算时间:

In [1]: import dask.dataframe as dd
   ...: df = dd.demo.daily_stock('GOOG', '2008', '2010', freq='1s', random_state=1234)
   ...: 

In [2]: %time df.close.mean().compute()
CPU times: user 7.73 s, sys: 1.15 s, total: 8.88 s
Wall time: 3.5 s
Out[2]: 452.30345234893554

In [3]: %time df = df.rename(columns={col: col.upper() for col in df.columns}); df.CLOSE.mean().compute()
CPU times: user 8.06 s, sys: 1.21 s, total: 9.27 s
Wall time: 3.81 s

In [4]: df.to_parquet('df', compression='GZIP')
   ...: df = dd.read_parquet('df')
   ...: 

In [5]: %time df.CLOSE.mean().compute()
CPU …

据我了解，镶木地板文件具有列的最小/最大统计信息。我的问题是如何在不读取整个文件的情况下使用 python 读取这些统计信息？

如果有帮助，我也有_common_metadata和_metadata文件。

我的具体问题是获取此文件系统中每个证券交易所分区的最大日期（每个年份分区包含多个具有日期列的镶木地板文件）：

C:.
?   _common_metadata
?   _metadata
????source=NASDAQ
?   ????year=2017
?   ????year=2018
????source=London_Stock_Exchange
?   ????year=2014
?   ????year=2015
????source=Japan_Exchange_Group
?   ????year=2017
?   ????year=2018
????source=Euronext
    ????year=2017
    ????year=2018

我正进入（状态：

KilledWorker: ("('from_pandas-1445321946b8a22fc0ada720fb002544', 4)", 'tcp://127.0.0.1:45940')

我已经阅读了关于后一个错误消息的解释，但这与堆栈跟踪顶部的错误消息一起出现令人困惑：

distributed.utils - 错误 - Worker 已经存在 tcp://127.0.0.1:35780

实际错误通过管道传输到Jupyter notebook为我的笔记本运行命令的终端：

ModuleNotFoundError：没有名为“_cython_magic_faba6120a194ab58ae9efd1da474433f”的模块

所以我将自己研究如何解决这个问题，现在我在我的案例中发现了详细的错误。关于这种特殊配置的精确提示会很好，但我想将所有 cython 代码提取到笔记本外部的 python 代码中更明智，而不是敲击了解 cython 魔术命令？

我正在尝试使用 command 从 dask.dataframe 中仅选择一行x.loc[0].compute()。它返回 4 行，所有行都具有index=0. 我试过了reset_index，但index=0重置后仍然会有 4 行。（我想我确实正确重置了，因为我做到了reset_index(drop=False)并且我可以在新列中看到原始索引）。

我阅读了dask.dataframe文档，它说，index=0由于 dask 如何构造块数据，可能会有不止一行的内容。

所以，如果我真的只想要一行index=0用于子集化，我该怎么做？

我试图将比 VRAM 多的数据传递到我的 GPU 中，这导致了以下错误。 CudaAPIError: Call to cuMemAlloc results in CUDA_ERROR_OUT_OF_MEMORY

我创建了这个代码来重现这个问题：

from numba import cuda
import numpy as np


@cuda.jit()
def addingNumbers (big_array, big_array2, save_array):
    i = cuda.grid(1)
    if i < big_array.shape[0]:
        for j in range (big_array.shape[1]):
            save_array[i][j] = big_array[i][j] * big_array2[i][j]



big_array = np.random.random_sample((1000000, 500))
big_array2  = np.random.random_sample((1000000, 500))
save_array = np.zeros(shape=(1000000, 500))


arraysize = 1000000
threadsperblock = 64
blockspergrid = (arraysize + (threadsperblock - 1))


d_big_array = cuda.to_device(big_array)
d_big_array2 = cuda.to_device(big_array2)
d_save_array = cuda.to_device(save_array)

addingNumbers[blockspergrid, threadsperblock](d_big_array, …