安利大家一個Python巨量資料分析神器

2020-12-30 18:00:18
欄目介紹一個巨量資料分析神器

推薦(免費):

對於Pandas執行速度的提升方法,之前已經介紹過很多回了,裡面經常提及Dask,很多朋友沒接觸過可能不太瞭解,今天就推薦一下這個神器。

1、什麼是Dask?

PandasNumpy大家都不陌生了,程式碼執行後資料都載入到RAM中,如果資料集特別大,我們就會看到記憶體飆升。但有時要處理的資料並不適合RAM,這時候Dask來了。

Dask是開源免費的。它是與其他社群專案(如Numpy,Pandas和Scikit-Learn)協調開發的。

官方:https://dask.org/

Dask支援PandasDataFrameNumpyArray的資料結構,並且既可在本地計算機上執行,也可以擴充套件到在叢集上執行。

基本上,只要編寫一次程式碼,使用普通的Pythonic語法,就可在本地執行或部署到多節點叢集上。這本身就是一個很牛逼的功能了,但這還不是最牛逼的。

我覺得Dask的最牛逼的功能是:它相容大部分我們已經在用的工具,並且只需改動少量的程式碼,就可以利用自己筆記型電腦上已有的處理能力並行執行程式碼。而並行處理資料就意味著更少的執行時間,更少的等待時間和更多的分析時間。

下面這個就是Dask進行資料處理的大致流程。
2f4ca94ca802585132464ae7a20a3a8.png

2、Dask支援哪些現有工具?

這一點也是我比較看中的,因為Dask可以與Python資料處理和建模的庫包相容,沿用庫包的API,這對於Python使用者來說學習成本是極低的。而像HadoopSpark這種巨量資料處理是有很高的學習門檻和時間成本的。

目前,Dask可支援pandasNumpySklearnXGBoostXArrayRAPIDS等等,光是這幾項我覺得就足夠用了,至少對於常用的資料處理、建模分析是完全覆蓋得掉的。
a4d1fd2bd472fee604a99f03976a305.png

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的資料,這些集合有DataFrameBagsArrays。這些集合型別中的每一個都能夠使用在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

下面說一下DaskDelay 功能,非常強大。

Dask.delayed是一種並行化現有程式碼的簡單而強大的方法。之所以被叫做delayed是因為,它沒有立即計算出結果,而是將要作為任務計算的結果記錄在一個圖形中,稍後將在並行硬體上執行。

有時問題用已有的dask.arraydask.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函數可修飾incdouble這些函數,以便它們可延遲執行,而不是立即執行函數,它將函數及其引數放入計算任務圖中。

我們簡單修改程式碼,用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)

程式碼執行後incdoubleaddsum都還沒有發生,而是生成一個計算的任務圖交給了total。然後我們用visualizatize看下任務圖。

total.visualize()

d538c247ad8ab6ccd08d350e5b4c405.png

上圖明顯看到了並行的可能性,所以毫不猶豫,使用compute進行平行計算,這時才完成了計算。

>>> total.compute()
45

由於資料集較小無法比較時間,這裡只介紹下使用方法,具體可自己動手實踐下。

Sklearn機器學習

關於機器學習的並行化執行,由於內容較多,東哥會在另一篇文章展開。這裡簡單說下一下dask-learn

dask-learn專案是與Sklearn開發人員共同作業完成的。現在可實現並行化有Scikit-learnPipelineGridsearchCVRandomSearchCV以及這些的變體,它們可以更好地處理巢狀的並行操作。

因此,如果你將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')

28295caa28bee6b9ee03a735ce7d68f.png

5、總結

以上就是Dask的簡單介紹,Dask的功能是非常強大的,且說明檔案也非常全,既有範例又有解釋。感興趣的朋友可以自行去官網或者GitHub學習,東哥下次分享使用Dask進行機器學習的一些範例。

以上就是安利大家一個Python巨量資料分析神器的詳細內容,更多請關注TW511.COM其它相關文章!