关于大数据:安利一个Python大数据分析神器

作者：东哥腾飞
首发于公众号：Python数据迷信

对于Pandas运行速度的晋升办法，之前曾经介绍过很多回了，外面常常提及Dask，很多敌人没接触过可能不太理解，明天就举荐一下这个神器。

1、什么是Dask？

Pandas和Numpy大家都不生疏了，代码运行后数据都加载到RAM中，如果数据集特地大，咱们就会看到内存飙升。但有时要解决的数据并不适宜RAM，这时候Dask来了。

Dask是开源收费的。它是与其余社区我的项目（如Numpy，Pandas和Scikit-Learn）协调开发的。

官网：https://dask.org/

Dask反对Pandas的DataFrame和NumpyArray的数据结构，并且既可在本地计算机上运行，也能够扩大到在集群上运行。

基本上，只有编写一次代码，应用一般的Pythonic语法，就可在本地运行或部署到多节点集群上。这自身就是一个很牛逼的性能了，但这还不是最牛逼的。

我感觉Dask的最牛逼的性能是：它兼容大部分咱们曾经在用的工具，并且只需改变大量的代码，就能够利用本人笔记本电脑上已有的解决能力并行运行代码。而并行处理数据就意味着更少的执行工夫，更少的等待时间和更多的剖析工夫。

上面这个就是Dask进行数据处理的大抵流程。

2、Dask反对哪些现有工具？

这一点也是我比拟看中的，因为Dask能够与Python数据处理和建模的库包兼容，沿用库包的API，这对于Python使用者来说学习老本是极低的。而像Hadoop、Spark这种大数据处理是有很高的学习门槛和工夫老本的。

目前，Dask可反对pandas、Numpy、Sklearn、XGBoost、XArray、RAPIDS等等，光是这几项我感觉就足够用了，至多对于罕用的数据处理、建模剖析是齐全笼罩得掉的。

3、Dask装置

能够应用 conda 或者 pip，或从源代码装置dask 。

conda install dask

因为dask有很多依赖，所以为了疾速装置也可用上面代码，将装置运行Dask所需的起码依赖关系集。

conda install dask-core

再有就是通过源来装置。

git clone https://github.com/dask/dask.git
cd dask
python -m pip install .

4、Dask如何应用？

Numpy、pandas

Dask引入了3个并行汇合，它们能够存储大于RAM的数据，这些汇合有DataFrame、Bags、Arrays。这些汇合类型中的每一个都可能应用在RAM和硬盘之间分区的数据，以及散布在群集中多个节点上的数据。

Dask的应用是十分清晰的，如果你应用NumPy数组，就从Dask数组开始，如果你应用Pandas DataFrame，就从Dask DataFrame开始，依此类推。

import dask.array as da
x = da.random.uniform(low=0, high=10, size=(10000, 10000),  # normal numpy code
                      chunks=(1000, 1000))  # break into chunks of size 1000x1000

y = x + x.T - x.mean(axis=0)  # Use normal syntax for high level algorithms

# DataFrames
import dask.dataframe as dd
df = dd.read_csv('2018-*-*.csv', parse_dates='timestamp',  # normal Pandas code
                 blocksize=64000000)  # break text into 64MB chunks

s = df.groupby('name').balance.mean()  # Use normal syntax for high level algorithms

# Bags / lists
import dask.bag as db
b = db.read_text('*.json').map(json.loads)
total = (b.filter(lambda d: d['name'] == 'Alice')
          .map(lambda d: d['balance'])
          .sum())

这些高级接口在稍微变动的状况下复制了标准接口。对于原始我的项目中的大部分API，这些接口会主动为咱们并行处理较大的数据集，实现上不是很简单，对照Dask的doc文档即可一步步实现。

Delayed

上面说一下Dask的 Delay 性能，十分弱小。

Dask.delayed是一种并行化现有代码的简略而弱小的办法。之所以被叫做delayed是因为，它没有立刻计算出后果，而是将要作为工作计算的后果记录在一个图形中，稍后将在并行硬件上运行。

有时问题用已有的dask.array或dask.dataframe可能都不适宜，在这些状况下，咱们能够应用更简略的dask.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)
45

下面代码在单个线程中按程序运行。然而，咱们看到其中很多能够并行执行。Dask delayed函数可润饰inc、double这些函数，以便它们可提早运行，而不是立刻执行函数，它将函数及其参数放入计算工作图中。

咱们简略批改代码，用delayed函数包装一下。

import dask

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

total = dask.delayed(sum)(output)

代码运行后inc、double、add和sum都还没有产生，而是生成一个计算的工作图交给了total。而后咱们用visualizatize看下工作图。

total.visualize()  

上图显著看到了并行的可能性，所以毫不犹豫，应用compute进行并行计算，这时才实现了计算。

>>> total.compute()
45

因为数据集较小无奈比拟工夫，这里只介绍下应用办法，具体可本人入手实际下。

Sklearn机器学习

对于机器学习的并行化执行，因为内容较多，东哥会在另一篇文章开展。这里简略说下一下dask-learn。

dask-learn我的项目是与Sklearn开发人员合作实现的。当初可实现并行化有Scikit-learn的Pipeline、GridsearchCV和RandomSearchCV以及这些的变体，它们能够更好地解决嵌套的并行操作。

因而，如果你将sklearn替换为dklearn，那么速度将会晋升很多。

# from sklearn.grid_search import GridSearchCV
  from dklearn.grid_search import GridSearchCV
# from sklearn.pipeline import Pipeline
  from dklearn.pipeline import Pipeline
上面是一个应用Pipeline的示例，其中利用了PCA和逻辑回归。
from sklearn.datasets import make_classification

X, y = make_classification(n_samples=10000,
                           n_features=500,
                           n_classes=2,
                           n_redundant=250,
                           random_state=42)

from sklearn import linear_model, decomposition
from sklearn.pipeline import Pipeline
from dklearn.pipeline import Pipeline

logistic = linear_model.LogisticRegression()
pca = decomposition.PCA()
pipe = Pipeline(steps=[('pca', pca),
                       ('logistic', logistic)])


grid = dict(pca__n_components=[50, 100, 150, 250],
            logistic__C=[1e-4, 1.0, 10, 1e4],
            logistic__penalty=['l1', 'l2'])

# from sklearn.grid_search import GridSearchCV
from dklearn.grid_search import GridSearchCV

estimator = GridSearchCV(pipe, grid)

estimator.fit(X, y)

后果是：sklearn会在40秒钟左右执行此计算，而dask-learn替代品大概须要10秒钟。
另外，如果增加以下代码能够连贯到集群，通过Client能够展现整个计算过程的dashboard，由Bokeh实现。

from dask.distributed import Client
c = Client('scheduler-address:8786')

5、总结

以上就是Dask的简略介绍，Dask的性能是十分弱小的，且阐明文档也十分全，既有示例又有解释。感兴趣的敌人能够自行去官网或者GitHub学习，东哥下次分享应用Dask进行机器学习的一些实例。

原创不易，感觉不错点个赞。

欢送关注我的集体公众号：Python数据迷信

数据迷信学习网站：datadeepin