本文的目標(biāo)受眾：

1. 想要對(duì)大量數(shù)據(jù)集執(zhí)行 Pandas/NumPy 操作的人。
2. 希望使用Python在大數(shù)據(jù)上執(zhí)行機(jī)器學(xué)習(xí)任務(wù)的人。

本文將使用 .csv 格式的文件來演示 python 的各種操作，其他格式如數(shù)組、文本文件等也是如此。

為什么我們不能將 pandas 用于大型機(jī)器學(xué)習(xí)數(shù)據(jù)集呢？

我們知道 Pandas 使用計(jì)算機(jī)內(nèi)存 (RAM) 來加載您的機(jī)器學(xué)習(xí)數(shù)據(jù)集，但是，如果您的計(jì)算機(jī)有8 GB 的內(nèi)存 (RAM)，那么為什么 pandas 仍然無法加載 2 GB 的數(shù)據(jù)集呢？原因是使用 Pandas 加載 2 GB 文件不僅需要 2 GB RAM，還需要更多內(nèi)存，因?yàn)榭們?nèi)存需求取決于數(shù)據(jù)集的大小以及您將在該數(shù)據(jù)集上執(zhí)行的操作。

以下是加載到計(jì)算機(jī)內(nèi)存中的不同大小的數(shù)據(jù)集的快速比較：

此外，Pandas只使用操作系統(tǒng)的一個(gè)內(nèi)核，這使得處理速度很慢。換句話說，我們可以說pandas不支持并行(將一個(gè)問題分解成更小的任務(wù))。

假設(shè)電腦有 4 個(gè)內(nèi)核，下圖是加載 CSV 文件的時(shí)候 pandas 使用的內(nèi)核數(shù)：

普遍不使用 pandas 處理大型機(jī)器學(xué)習(xí)數(shù)據(jù)集的主要原因有以下兩點(diǎn)，一是計(jì)算機(jī)內(nèi)存使用量，二是缺乏并行性。在 NumPy 和 Scikit-learn中，對(duì)于大數(shù)據(jù)集也面臨同樣的問題。

為了解決這兩個(gè)問題，可以使用名為Dask的python庫，它能夠使我們?cè)诖笮蛿?shù)據(jù)集上執(zhí)行pandas、NumPy和ML等各種操作。

Dask是如何工作的？

Dask是在分區(qū)中加載你的數(shù)據(jù)集，而pandas通常是將整個(gè)機(jī)器學(xué)習(xí)數(shù)據(jù)集作為一個(gè)dataframe。在Dask中，數(shù)據(jù)集的每個(gè)分區(qū)都被認(rèn)為是一個(gè)pandas dataframe。

Dask 一次加載一個(gè)分區(qū)，因此您不必?fù)?dān)心出現(xiàn)內(nèi)存分配錯(cuò)誤問題。

以下是使用 dask 在計(jì)算機(jī)內(nèi)存中加載不同大小的機(jī)器學(xué)習(xí)數(shù)據(jù)集的比較：

Dask 解決了并行性問題，因?yàn)樗鼘?shù)據(jù)拆分為多個(gè)分區(qū)，每個(gè)分區(qū)使用一個(gè)單獨(dú)的內(nèi)核，這使得數(shù)據(jù)集上的計(jì)算更快。

假設(shè)電腦有 4 個(gè)內(nèi)核，以下是 dask 在加載 5 GB csv 文件時(shí)的方式：

要使用 dask 庫，您可以使用以下命令進(jìn)行安裝：

pip install dask

Dask 有幾個(gè)模塊，如dask.array、dask.dataframe 和 dask.distributed，只有在您分別安裝了相應(yīng)的庫（如 NumPy、pandas 和 Tornado）后才能工作。

如何使用 dask 處理大型 CSV 文件？

dask.dataframe 用于處理大型 csv 文件，首先我嘗試使用 pandas 導(dǎo)入大小為 8 GB 的數(shù)據(jù)集。

import pandas as pd
df = pd.read_csv(“data.csv”)

它在我的 16 GB 內(nèi)存筆記本電腦中引發(fā)了內(nèi)存分配錯(cuò)誤。

現(xiàn)在，嘗試使用 dask.dataframe 導(dǎo)入相同的 8 GB 數(shù)據(jù)

dask 只用了一秒鐘就將整個(gè) 8 GB 文件加載到 ddf 變量中。

讓我們看看 ddf 變量的輸出。

如您所見，執(zhí)行時(shí)間為 0.5 秒，這里顯示已劃分為 119 個(gè)分區(qū)。

您還可以使用以下方法檢查數(shù)據(jù)幀的分區(qū)數(shù)：

默認(rèn)情況下，dask 將我的 8 GB CSV 文件加載到 119 個(gè)分區(qū)（每個(gè)分區(qū)大小為 64MB），這是根據(jù)可用的物理內(nèi)存和電腦的內(nèi)核數(shù)來完成的。

還可以在加載 CSV 文件時(shí)使用 blocksize 參數(shù)指定我自己的分區(qū)數(shù)。

現(xiàn)在指定了一個(gè)字符串值為 400MB 的 blocksize 參數(shù)，這使得每個(gè)分區(qū)大小為 400 MB，讓我們看看有多少個(gè)分區(qū)

關(guān)鍵點(diǎn)：使用 Dask DataFrames 時(shí)，一個(gè)好的經(jīng)驗(yàn)法則是將分區(qū)保持在 100MB 以下。

使用以下方法可調(diào)用dataframe的特定分區(qū)：

也可通過使用負(fù)索引來調(diào)用最后一個(gè)分區(qū)，就像我們?cè)谡{(diào)用列表的最后一個(gè)元素時(shí)所做的那樣。

讓我們看看數(shù)據(jù)集的形狀：

您可以使用 len() 檢查數(shù)據(jù)集的行數(shù)：

Dask 已經(jīng)包含了示例數(shù)據(jù)集。我將使用時(shí)間序列數(shù)據(jù)向您展示 dask 如何對(duì)數(shù)據(jù)集執(zhí)行數(shù)學(xué)運(yùn)算。

導(dǎo)入dask.datasets后，ddf_20y 加載了從 2000 年 1 月 1 日到 2021 年 12 月 31 日的時(shí)間序列數(shù)據(jù)。

讓我們看看我們的時(shí)間序列數(shù)據(jù)的分區(qū)數(shù)。

20 年的時(shí)間序列數(shù)據(jù)分布在 8035 個(gè)分區(qū)中。

在 pandas 中，我們使用 head 打印數(shù)據(jù)集的前幾行，dask 也是這樣。

讓我們計(jì)算一下 id 列的平均值。

dask不會(huì)打印dataframe的總行數(shù)，因?yàn)樗褂枚栊杂?jì)算(直到需要時(shí)才顯示輸出)。為了顯示輸出，我們可以使用compute方法。

假設(shè)我想對(duì)數(shù)據(jù)集的每一列進(jìn)行歸一化(將值轉(zhuǎn)換為0到1之間)，Python代碼如下：

循環(huán)遍歷列，找到每列的最小值和最大值，并使用簡(jiǎn)單的數(shù)學(xué)公式對(duì)這些列進(jìn)行歸一化。

關(guān)鍵點(diǎn)：在我們的歸一化示例中，不要認(rèn)為會(huì)發(fā)生實(shí)際的數(shù)值計(jì)算，它只是惰性求值（在需要之前永遠(yuǎn)不會(huì)向您顯示輸出）。

為什么要使用 Dask 數(shù)組？

Dask 將數(shù)組分成小塊，其中每個(gè)塊都是一個(gè) NumPy 數(shù)組。

dask.arrays 用于處理大數(shù)組，以下Python代碼使用 dask 創(chuàng)建了一個(gè) 10000 x 10000 的數(shù)組并將其存儲(chǔ)在 x 變量中。

調(diào)用該 x 變量會(huì)產(chǎn)生有關(guān)數(shù)組的各種信息。

查看數(shù)組的特定元素

對(duì)dask 數(shù)組進(jìn)行數(shù)學(xué)運(yùn)算的Python示例：

正如您所看到的，由于延遲執(zhí)行，它不會(huì)向您顯示輸出。我們可以使用compute來顯示輸出：

dask 數(shù)組支持大多數(shù) NumPy 接口，如下所示：

• 數(shù)學(xué)運(yùn)算：+, *, exp, log, ...
• sum(), mean(), std(), sum(axis=0), ...
• 張量/點(diǎn)積/矩陣乘法：tensordot
• 重新排序/轉(zhuǎn)置：transpose
• 切片：x[:100, 500:100:-2]
• 使用列表或 NumPy 數(shù)組進(jìn)行索引：x[:, [10, 1, 5]]
• 線性代數(shù)：svd、qr、solve、solve_triangular、lstsq

但是，Dask Array 并沒有實(shí)現(xiàn)完整 NumPy 接口。

你可以從他們的官方文檔中了解更多關(guān)于 dask.arrays 的信息。

什么是Dask Persist？

假設(shè)您想對(duì)機(jī)器學(xué)習(xí)數(shù)據(jù)集執(zhí)行一些耗時(shí)的操作，您可以將數(shù)據(jù)集持久化到內(nèi)存中，從而使數(shù)學(xué)運(yùn)算運(yùn)行得更快。

從 dask.datasets 導(dǎo)入了時(shí)間序列數(shù)據(jù)

讓我們?nèi)?shù)據(jù)集的一個(gè)子集并計(jì)算該子集的總行數(shù)。

計(jì)算總行數(shù)需要 27 秒。

我們現(xiàn)在使用 persist 方法：

持久化我們的子集總共花了 2 分鐘，現(xiàn)在讓我們計(jì)算總行數(shù)。

同樣，我們可以對(duì)持久化數(shù)據(jù)集執(zhí)行其他操作以減少計(jì)算時(shí)間。

persist應(yīng)用場(chǎng)景：

• 數(shù)據(jù)量大
• 獲取數(shù)據(jù)的一個(gè)子集
• 對(duì)子集應(yīng)用不同的操作

為什么選擇 Dask ML？

Dask ML有助于在大型數(shù)據(jù)集上使用流行的Python機(jī)器學(xué)習(xí)庫(如Scikit learn等)來應(yīng)用ML（機(jī)器學(xué)習(xí)）算法。

什么時(shí)候應(yīng)該使用 dask ML？

• 數(shù)據(jù)不大（或適合 RAM），但訓(xùn)練的機(jī)器學(xué)習(xí)模型需要大量超參數(shù)，并且調(diào)優(yōu)或集成技術(shù)需要大量時(shí)間。
• 數(shù)據(jù)量很大。

正如你所看到的，隨著模型大小的增加，例如，制作一個(gè)具有大量超參數(shù)的復(fù)雜模型，它會(huì)引起計(jì)算邊界的問題，而如果數(shù)據(jù)大小增加，它會(huì)引起內(nèi)存分配錯(cuò)誤。因此，在這兩種情況下（紅色陰影區(qū)域）我們都使用 Dask 來解決這些問題。

如官方文檔中所述，dask ml 庫用例：

• 對(duì)于內(nèi)存問題，只需使用 scikit-learn（或其他ML 庫）。
• 對(duì)于大型模型，使用 dask_ml.joblib 和scikit-learn estimators。
• 對(duì)于大型數(shù)據(jù)集，使用 dask_ml estimators。

讓我們看一下 Dask.distributed 的架構(gòu)：

Dask 讓您能夠在計(jì)算機(jī)集群上運(yùn)行任務(wù)。在 dask.distributed 中，只要您分配任務(wù)，它就會(huì)立即開始執(zhí)行。

簡(jiǎn)單地說，client就是提交任務(wù)的你，執(zhí)行任務(wù)的是Worker，調(diào)度器則執(zhí)行兩者之間通信。

python -m pip install dask distributed –upgrade

如果您使用的是單臺(tái)機(jī)器，那么就可以通過以下方式創(chuàng)建一個(gè)具有4個(gè)worker的dask集群

如果需要dashboard，可以安裝bokeh，安裝bokeh的命令如下：

pip install bokeh

就像我們從 dask.distributed 創(chuàng)建客戶端一樣，我們也可以從 dask.distributed 創(chuàng)建調(diào)度程序。

要使用 dask ML 庫，您必須使用以下命令安裝它：

pip install dask-ml

我們將使用 Scikit-learn 庫來演示 dask-ml 。

假設(shè)我們使用 Grid_Search 方法，我們通常使用如下Python代碼

使用 dask.distributed 創(chuàng)建一個(gè)集群：

要使用集群擬合 scikit-learn 模型，我們只需要使用 joblib。