Python 大规模数据存储与读取、并行计算：Dask库简述

拼搏现实的明天。 2022-03-25 19:34 389阅读 0赞

[本文转自：https://blog.csdn.net/sinat\_26917383/article/details/78044437][https_blog.csdn.net_sinat_26917383_article_details_78044437]

数据结构与pandas非常相似，比较容易理解。

*  原文文档：[http://dask.pydata.org/en/latest/index.html][http_dask.pydata.org_en_latest_index.html]

github：[https://github.com/dask][https_github.com_dask]

dask的内容很多，挑一些我比较看好的内容着重点一下。

# 一、数据读取与存储 #

先来看看dask能读入哪些内容：  
![这里写图片描述][SouthEast]

1、csv

# pandas
    import pandas as pd                    
    df = pd.read_csv('2015-01-01.csv')      
    df.groupby(df.user_id).value.mean()     
    
    #dask
     import dask.dataframe as dd
     df = dd.read_csv('2015-*-*.csv')
     df.groupby(df.user_id).value.mean().compute()

非常相似，除了.compute()  
.  
2、Dask Array读取hdf5

import numpy as np                       import dask.array as da
    f = h5py.File('myfile.hdf5')             f = h5py.File('myfile.hdf5')
    x = np.array(f['/small-data'])           x = da.from_array(f['/big-data'],
                                                               chunks=(1000, 1000))
    x - x.mean(axis=1)                       x - x.mean(axis=1).compute()

左是Pandas，右边是dask

3、Dask Bag

import dask.bag as db  
b = db.read\_text('2015-\*-\*.json.gz').map(json.loads)  
b.pluck('name').frequencies().topk(10, lambda pair: pair\[1\]).compute()

读取大规模json文件，几亿都很easy

>>> b = db.read\_text('myfile.txt')  
>>> b = db.read\_text(\['myfile.1.txt', 'myfile.2.txt', ...\])  
>>> b = db.read\_text('myfile.\*.txt')

读取txt

>>> import dask.bag as db  
>>> b = db.from\_sequence(\[\{'name': 'Alice', 'balance': 100\},  
... \{'name': 'Bob', 'balance': 200\},  
... \{'name': 'Charlie', 'balance': 300\}\],  
... npartitions=2)  
>>> df = b.to\_dataframe()

变为dataframe格式的内容

4、Dask Delayed 并行计算

from dask import delayed  
L = \[\]  
for fn in filenames: \# Use for loops to build up computation  
data = delayed(load)(fn) \# Delay execution of function  
L.append(delayed(process)(data)) \# Build connections between variables

result = delayed(summarize)(L)  
result.compute()

5、concurrent.futures自定义任务

from dask.distributed import Client  
client = Client('scheduler:port')

futures = \[\]  
for fn in filenames:  
future = client.submit(load, fn)  
futures.append(future)

summary = client.submit(summarize, futures)  
summary.result()

二、Delayed 并行计算模块

一个先行例子，本来的案例：

def inc(x):  
return x + 1

def double(x):  
return x + 2

def add(x, y):  
return x + y

data = \[1, 2, 3, 4, 5\]

output = \[\]  
for x in data:  
a = inc(x)  
b = double(x)  
c = add(a, b)  
output.append(c)

total = sum(output)

再来看看用delay加速的：  
这里写图片描述

from dask import delayed

output = \[\]  
for x in data:  
a = delayed(inc)(x)  
b = delayed(double)(x)  
c = delayed(add)(a, b)  
output.append(c)

total = delayed(sum)(output)

还可以将计算流程可视化:

total.visualize() \# see image to the right

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTE1OTY0NTU_size_16_color_FFFFFF_t_70][]

三、和SKLearn结合的并行算法

广义回归GLM：https://github.com/dask/dask-glm  
tensorflow深度学习库：Dask-Tensorflow

以XGBoost为例，官方：https://github.com/dask/dask-xgboost  
来看一个案例code  
.  
1、加载数据

import dask.dataframe as dd

\# Subset of the columns to use  
cols = \['Year', 'Month', 'DayOfWeek', 'Distance',  
'DepDelay', 'CRSDepTime', 'UniqueCarrier', 'Origin', 'Dest'\]

\# Create the dataframe  
df = dd.read\_csv('s3://dask-data/airline-data/20\*.csv', usecols=cols,  
storage\_options=\{'anon': True\})

df = df.sample(frac=0.2) \# we blow out ram otherwise

is\_delayed = (df.DepDelay.fillna(16) > 15)

df\['CRSDepTime'\] = df\['CRSDepTime'\].clip(upper=2399)  
del df\['DepDelay'\]

df, is\_delayed = persist(df, is\_delayed)  
progress(df, is\_delayed)

2、One hot encode编码

df2 = dd.get\_dummies(df.categorize()).persist()

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTE1OTY0NTU_size_16_color_FFFFFF_t_70 1][]

这里写图片描述  
.  
3、准备训练集和测试集 + 训练

data\_train, data\_test = df2.random\_split(\[0.9, 0.1\],  
random\_state=1234)  
labels\_train, labels\_test = is\_delayed.random\_split(\[0.9, 0.1\],  
random\_state=1234)

训练

import dask\_xgboost as dxgb

params = \{'objective': 'binary:logistic', 'nround': 1000,  
'max\_depth': 16, 'eta': 0.01, 'subsample': 0.5,  
'min\_child\_weight': 1\}

bst = dxgb.train(client, params, data\_train, labels\_train)  
bst

4、预测

\# Use normal XGBoost model with normal Pandas  
import xgboost as xgb  
dtest = xgb.DMatrix(data\_test.head())  
bst.predict(dtest)

predictions = dxgb.predict(client, bst, data\_test).persist()  
predictions.head()

.  
5、模型评估

from sklearn.metrics import roc\_auc\_score, roc\_curve  
print(roc\_auc\_score(labels\_test.compute(),  
predictions.compute()))  
import matplotlib.pyplot as plt  
%matplotlib inline

fpr, tpr, \_ = roc\_curve(labels\_test.compute(), predictions.compute())  
\# Taken from http://scikit-learn.org/stable/auto\_examples/model\_selection/plot\_roc.html\#sphx-glr-auto-examples-model-selection-plot-roc-py  
plt.figure(figsize=(8, 8))  
lw = 2  
plt.plot(fpr, tpr, color='darkorange', lw=lw, label='ROC curve')  
plt.plot(\[0, 1\], \[0, 1\], color='navy', lw=lw, linestyle='--')  
plt.xlim(\[0.0, 1.0\])  
plt.ylim(\[0.0, 1.05\])  
plt.xlabel('False Positive Rate')  
plt.ylabel('True Positive Rate')  
plt.title('Receiver operating characteristic example')  
plt.legend(loc="lower right")  
plt.show()

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTE1OTY0NTU_size_16_color_FFFFFF_t_70 2][]  
.  
四、计算流程可视化部分——Dask.array

来源：https://gist.github.com/mrocklin/b61f795004ec0a70e43de350e453e97e

import numpy as np  
import dask.array as da  
x = da.ones(15, chunks=(5,))  
x.visualize('dask.svg')  
![20190121205023532.png][]

(x + 1).sum().visualize('dask.svg')

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTE1OTY0NTU_size_16_color_FFFFFF_t_70 3][]

来一个二维模块的：

x = da.ones((15, 15), chunks=(5, 5))
    x.visualize('dask.svg')
    (x.dot(x.T + 1) - x.mean(axis=0)).std().visualize('dask.svg')

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTE1OTY0NTU_size_16_color_FFFFFF_t_70 4][]

\---------------------  
作者：悟乙己  
来源：CSDN  
原文：https://blog.csdn.net/sinat\_26917383/article/details/78044437  
版权声明：本文为博主原创文章，转载请附上博文链接！

[https_blog.csdn.net_sinat_26917383_article_details_78044437]: https://blog.csdn.net/sinat_26917383/article/details/78044437
[http_dask.pydata.org_en_latest_index.html]: http://dask.pydata.org/en/latest/index.html
[https_github.com_dask]: https://github.com/dask
[SouthEast]: /images/20220325/2281ca781dcd4abca787754d6fd5b4ba.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTE1OTY0NTU_size_16_color_FFFFFF_t_70]: /images/20220325/6a488721bb8d4273b36218bd3310b701.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTE1OTY0NTU_size_16_color_FFFFFF_t_70 1]: /images/20220325/ddd8c920e3e7415e8d8604ba955da59b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTE1OTY0NTU_size_16_color_FFFFFF_t_70 2]: /images/20220325/fae09d611f4546a7b3bfc2e522fef28e.png
[20190121205023532.png]: /images/20220325/721e0e650f894b45a8c732d9b8759657.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTE1OTY0NTU_size_16_color_FFFFFF_t_70 3]: /images/20220325/a011ba6b285e41d3abd92e41ac7dff90.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTE1OTY0NTU_size_16_color_FFFFFF_t_70 4]: /images/20220325/e718f37961e143a09ce1ca7f3fe31bc6.png