在數(shù)據(jù)科學(xué)世界，Python 是一個(gè)不可忽視的存在，且有愈演愈烈之勢(shì)。而其中主要的使用工具，包括 Numpy、Pandas 和 Scikit-learn 等。 Mars 在 MaxCompute 團(tuán)隊(duì)內(nèi)部誕生，本文將分享如何通過 Mars 讓 Numpy、pandas 和 scikit-learn 等數(shù)據(jù)科學(xué)的庫(kù)能夠并行和分布式執(zhí)行，并和 RAPIDS 平臺(tái)結(jié)合用 GPU 來加速數(shù)據(jù)科學(xué)。

Numpy

Numpy 是數(shù)值計(jì)算的基礎(chǔ)包，內(nèi)部提供了多維數(shù)組(ndarray)這樣一個(gè)數(shù)據(jù)結(jié)構(gòu)，用戶可以很方便地在任意維度上進(jìn)行數(shù)值計(jì)算。

??

我們舉一個(gè)蒙特卡洛方法求解 Pi 的例子。這背后的原理非常簡(jiǎn)單，現(xiàn)在我們有個(gè)半徑為1的圓和邊長(zhǎng)為2的正方形，他們的中心都在原點(diǎn)?，F(xiàn)在我們生成大量的均勻分布的點(diǎn)，讓這些點(diǎn)落在正方形內(nèi)，通過簡(jiǎn)單的推導(dǎo)，我們就可以知道，Pi 的值 = 落在圓內(nèi)的點(diǎn)的個(gè)數(shù) / 點(diǎn)的總數(shù) * 4。

這里要注意，就是隨機(jī)生成的點(diǎn)的個(gè)數(shù)越多，結(jié)果越精確。

用 Numpy 實(shí)現(xiàn)如下：

import numpy as np 
  
 N = 10 ** 7  # 1千萬個(gè)點(diǎn) 
  
 data = np.random.uniform(-1, 1, size=(N, 2))  # 生成1千萬個(gè)x軸和y軸都介于-1和1間的點(diǎn) 
 inside = (np.sqrt((data ** 2).sum(axis=1)) < 1).sum()  # 計(jì)算到原點(diǎn)的距離小于1的點(diǎn)的個(gè)數(shù) 
 pi = 4 * inside / N 
 print('pi: %.5f' % pi)

可以看到，用 Numpy 來進(jìn)行數(shù)值計(jì)算非常簡(jiǎn)單，只要寥寥數(shù)行代碼，而如果讀者習(xí)慣了 Numpy 這種面相數(shù)組的思維方式之后，無論是代碼的可讀性還是執(zhí)行效率都會(huì)有巨大提升。

pandas

pandas 是一個(gè)強(qiáng)大的數(shù)據(jù)分析和處理的工具，它其中包含了海量的 API 來幫助用戶在二維數(shù)據(jù)(DataFrame)上進(jìn)行分析和處理。

pandas 中的一個(gè)核心數(shù)據(jù)結(jié)構(gòu)就是 DataFrame，它可以簡(jiǎn)單理解成表數(shù)據(jù)，但不同的是，它在行和列上都包含索引(Index)，要注意這里不同于數(shù)據(jù)庫(kù)的索引的概念，它的索引可以這么理解：當(dāng)從行看 DataFrame 時(shí)，我們可以把 DataFrame 看成行索引到行數(shù)據(jù)的這么一個(gè)字典，通過行索引，可以很方便地選中一行數(shù)據(jù);列也同理。

我們拿 movielens 的數(shù)據(jù)集 作為簡(jiǎn)單的例子，來看 pandas 是如何使用的。這里我們用的是 Movielens 20M Dataset.

import pandas as pd 
  
 ratings = pd.read_csv('ml-20m/ratings.csv') 
 ratings.groupby('userId').agg({'rating': ['sum', 'mean', 'max', 'min']})

通過一行簡(jiǎn)單的 pandas.read_csv 就可以讀取 CSV 數(shù)據(jù)，接著按 userId 做分組聚合，求 rating 這列在每組的總和、平均、最大、最小值。

“食用“ pandas 的最佳方式，還是在 Jupyter notebook 里，以交互式的方式來分析數(shù)據(jù)，這種體驗(yàn)會(huì)讓你不由感嘆：人生苦短，我用 xx(??)

scikit-learn

scikit-learn 是一個(gè) Python 機(jī)器學(xué)習(xí)包，提供了大量機(jī)器學(xué)習(xí)算法，用戶不需要知道算法的細(xì)節(jié)，只要通過幾個(gè)簡(jiǎn)單的 high-level 接口就可以完成機(jī)器學(xué)習(xí)任務(wù)。當(dāng)然現(xiàn)在很多算法都使用深度學(xué)習(xí)，但 scikit-learn 依然能作為基礎(chǔ)機(jī)器學(xué)習(xí)庫(kù)來串聯(lián)整個(gè)流程。

我們以 K-最鄰近算法為例，來看看用 scikit-learn 如何完成這個(gè)任務(wù)。

import pandas as pd 
 from sklearn.neighbors import NearestNeighbors 
  
 df = pd.read_csv('data.csv')  # 輸入是 CSV 文件，包含 20萬個(gè)向量，每個(gè)向量10個(gè)元素 
 nn = NearestNeighbors(n_neighbors=10) 
 nn.fit(df) 
 neighbors = nn.kneighbors(df)

fit 接口就是 scikit-learn 里最常用的用來學(xué)習(xí)的接口?？梢钥吹秸麄€(gè)過程非常簡(jiǎn)單易懂。

Mars —— Numpy、pandas 和 scikit-learn 的并行和分布式加速器

Python 數(shù)據(jù)科學(xué)棧非常強(qiáng)大，但它們有如下幾個(gè)問題：

• 現(xiàn)在是多核時(shí)代，這幾個(gè)庫(kù)里鮮有操作能利用得上多核的能力。
• 隨著深度學(xué)習(xí)的流行，用來加速數(shù)據(jù)科學(xué)的新的硬件層出不窮，這其中最常見的就是 GPU，在深度學(xué)習(xí)前序流程中進(jìn)行數(shù)據(jù)處理，我們是不是也能用上 GPU 來加速呢?
• 這幾個(gè)庫(kù)的操作都是命令式的(imperative)，和命令式相對(duì)應(yīng)的就是聲明式(declarative)。命令式的更關(guān)心 how to do，每一個(gè)操作都會(huì)立即得到結(jié)果，方便對(duì)結(jié)果進(jìn)行探索，優(yōu)點(diǎn)是很靈活;缺點(diǎn)則是中間過程可能占用大量?jī)?nèi)存，不能及時(shí)釋放，而且每個(gè)操作之間就被割裂了，沒有辦法做算子融合來提升性能;那相對(duì)應(yīng)的聲明式就剛好相反，它更關(guān)心 what to do，它只關(guān)心結(jié)果是什么，中間怎么做并沒有這么關(guān)心，典型的聲明式像 SQL、TensorFlow 1.x，聲明式可以等用戶真正需要結(jié)果的時(shí)候才去執(zhí)行，也就是 lazy evaluation，這中間過程就可以做大量的優(yōu)化，因此性能上也會(huì)有更好的表現(xiàn)，缺點(diǎn)自然也就是命令式的優(yōu)點(diǎn)，它不夠靈活，調(diào)試起來比較困難。

為了解決這幾個(gè)問題，Mars 被我們開發(fā)出來，Mars 在 MaxCompute 團(tuán)隊(duì)內(nèi)部誕生，它的主要目標(biāo)就是讓 Numpy、pandas 和 scikit-learn 等數(shù)據(jù)科學(xué)的庫(kù)能夠并行和分布式執(zhí)行，充分利用多核和新的硬件。

Mars 的開發(fā)過程中，我們核心關(guān)注的幾點(diǎn)包括：

• 我們希望 Mars 足夠簡(jiǎn)單，只要會(huì)用 Numpy、pandas 或 scikit-learn 就會(huì)用 Mars。
• 避免重復(fù)造輪子，我們希望能利用到這些庫(kù)已有的成果，只需要能讓他們被調(diào)度到多核/多機(jī)上即可。
• 聲明式和命令式兼得，用戶可以在這兩者之間自由選擇，靈活度和性能兼而有之。
• 足夠健壯，生產(chǎn)可用，能應(yīng)付各種 failover 的情況。

當(dāng)然這些是我們的目標(biāo)，也是我們一直努力的方向。

Mars tensor：Numpy 的并行和分布式加速器

上面說過，我們的目標(biāo)之一是，只要會(huì)用 Numpy 等數(shù)據(jù)科學(xué)包，就會(huì)用 Mars。我們直接來看代碼，還是以蒙特卡洛為例。變成 Mars 的代碼是什么樣子呢?

import mars.tensor as mt 
  
 N = 10 ** 10 
  
 data = mt.random.uniform(-1, 1, size=(N, 2)) 
 inside = (mt.sqrt((data ** 2).sum(axis=1)) < 1).sum() 
 pi = (4 * inside / N).execute() 
 print('pi: %.5f' % pi)

可以看到，區(qū)別就只有兩處：import numpy as np 變成 import mars.tensor as mt ，后續(xù)的 np. 都變成 mt. ;pi 在打印之前調(diào)用了一下 .execute() 方法。

也就是默認(rèn)情況下，Mars 會(huì)按照聲明式的方式，代碼本身移植的代價(jià)極低，而在真正需要一個(gè)數(shù)據(jù)的時(shí)候，通過 .execute() 去觸發(fā)執(zhí)行。這樣能最大限度得優(yōu)化性能，以及減少中間過程內(nèi)存消耗。

這里，我們還將數(shù)據(jù)的規(guī)模擴(kuò)大了 1000 倍，來到了 100 億個(gè)點(diǎn)。之前 1/1000 的數(shù)據(jù)量的時(shí)候，在我的筆記本上需要 757ms;而現(xiàn)在數(shù)據(jù)擴(kuò)大一千倍，光 data 就需要 150G 的內(nèi)存，這用 Numpy 本身根本無法完成。而使用 Mars，計(jì)算時(shí)間只需要 3min 44s，而峰值內(nèi)存只需要 1G 左右。假設(shè)我們認(rèn)為內(nèi)存無限大，Numpy 需要的時(shí)間也就是之前的 1000 倍，大概是 12min 多，可以看到 Mars 充分利用了多核的能力，并且通過聲明式的方式，極大減少了中間內(nèi)存占用。

前面說到，我們?cè)噲D讓聲明式和命令式兼得，而使用命令式的風(fēng)格，只需要在代碼的開始配置一個(gè)選項(xiàng)即可。

import mars.tensor as mt 
 from mars.config import options 
  
 options.eager_mode = True  # 打開 eager mode 后，每一次調(diào)用都會(huì)立即執(zhí)行，行為和 Numpy 就完全一致 
  
 N = 10 ** 7 
  
 data = mt.random.uniform(-1, 1, size=(N, 2)) 
 inside = (mt.linalg.norm(data, axis=1) < 1).sum() 
 pi = 4 * inside / N  # 不需要調(diào)用 .execute() 了 
 print('pi: %.5f' % pi.fetch())  # 目前需要 fetch() 來轉(zhuǎn)成 float 類型，后續(xù)我們會(huì)加入自動(dòng)轉(zhuǎn)換

Mars DataFrame：pandas 的并行和分布式加速器

看過怎么樣輕松把 Numpy 代碼遷移到 Mars tensor ，想必讀者也知道怎么遷移 pandas 代碼了，同樣也只有兩個(gè)區(qū)別。我們還是以 movielens 的代碼為例。

import mars.dataframe as md 
  
 ratings = md.read_csv('ml-20m/ratings.csv') 
 ratings.groupby('userId').agg({'rating': ['sum', 'mean', 'max', 'min']}).execute()

Mars Learn：scikit-learn 的并行和分布式加速器

Mars Learn 也同理，這里就不做過多闡述了。但目前 Mars learn 支持的 scikit-learn 算法還不多，我們也在努力移植的過程中，這需要大量的人力和時(shí)間，歡迎感興趣的同學(xué)一起參與。

import mars.dataframe as md 
 from mars.learn.neighbors import NearestNeighbors 
  
 df = md.read_csv('data.csv')  # 輸入是 CSV 文件，包含 20萬個(gè)向量，每個(gè)向量10個(gè)元素 
 nn = NearestNeighbors(n_neighbors=10) 
 nn.fit(df)  # 這里 fit 的時(shí)候也會(huì)整體觸發(fā)執(zhí)行，因此機(jī)器學(xué)習(xí)的高層接口都是立即執(zhí)行的 
 neighbors = nn.kneighbors(df).fetch()  # kneighbors 也已經(jīng)觸發(fā)執(zhí)行，只需要 fetch 數(shù)據(jù)

這里要注意的是，對(duì)于機(jī)器學(xué)習(xí)的 fit、predict 等高層接口，Mars Learn 也會(huì)立即觸發(fā)執(zhí)行，以保證語(yǔ)義的正確性。

RAPIDS：GPU 上的數(shù)據(jù)科學(xué)

相信細(xì)心的觀眾已經(jīng)發(fā)現(xiàn)，GPU 好像沒有被提到。不要著急，這就要說到 RAPIDS。

在之前，雖然 CUDA 已經(jīng)將 GPU 編程的門檻降到相當(dāng)?shù)偷囊粋€(gè)程度了，但對(duì)于數(shù)據(jù)科學(xué)家們來說，在 GPU 上處理 Numpy、pandas 等能處理的數(shù)據(jù)無異于天方夜譚。幸運(yùn)的是，NVIDIA 開源了 RAPIDS 數(shù)據(jù)科學(xué)平臺(tái)，它和 Mars 的部分思想高度一致，即使用簡(jiǎn)單的 import 替換，就可以將 Numpy、pandas 和 scikit-learn 的代碼移植到 GPU 上。

??

其中，RAPIDS cuDF 用來加速 pandas，而 RAPIDS cuML 用來加速 scikit-learn。

對(duì)于 Numpy 來說，CuPy 已經(jīng)很好地支持用 GPU 來加速了，這樣 RAPIDS 也得以把重心放在數(shù)據(jù)科學(xué)的其他部分。

CuPy：用 GPU 加速 Numpy

還是蒙特卡洛求解 Pi。

import cupy as cp 
   
 N = 10 ** 7 
  
 data = cp.random.uniform(-1, 1, size=(N, 2)) 
 inside = (cp.sqrt((data ** 2).sum(axis=1)) < 1).sum() 
 pi = 4 * inside / N 
 print('pi: %.5f' % pi)

在我的測(cè)試中，它將 CPU 的 757ms，降到只有 36ms，提升超過 20 倍，可以說效果非常顯著。這正是得益于 GPU 非常適合計(jì)算密集型的任務(wù)。

RAPIDS cuDF：用 GPU 加速 pandas

將 import pandas as pd 替換成 import cudf，GPU 內(nèi)部如何并行，CUDA 編程這些概念，用戶都不再需要關(guān)心。

import cudf 
  
 ratings = cudf.read_csv('ml-20m/ratings.csv') 
 ratings.groupby('userId').agg({'rating': ['sum', 'mean', 'max', 'min']})

運(yùn)行時(shí)間從 CPU 上的 18s 提升到 GPU 上的 1.66s，提升超過 10 倍。

RAPIDS cuML：用 GPU 加速 scikit-learn

同樣是 k-最鄰近問題。

import cudf 
 from cuml.neighbors import NearestNeighbors 
  
 df = cudf.read_csv('data.csv') 
 nn = NearestNeighbors(n_neighbors=10) 
 nn.fit(df) 
 neighbors = nn.kneighbors(df)

運(yùn)行時(shí)間從 CPU 上 1min52s，提升到 GPU 上 17.8s。

Mars 和 RAPIDS 結(jié)合能帶來什么?

RAPIDS 將 Python 數(shù)據(jù)科學(xué)帶到了 GPU，極大地提升了數(shù)據(jù)科學(xué)的運(yùn)行效率。它們和 Numpy 等一樣，是命令式的。通過和 Mars 結(jié)合，中間過程將會(huì)使用更少的內(nèi)存，這使得數(shù)據(jù)處理量更大;Mars 也可以將計(jì)算分散到多機(jī)多卡，以提升數(shù)據(jù)規(guī)模和計(jì)算效率。

在 Mars 里使用 GPU 也很簡(jiǎn)單，只需要在對(duì)應(yīng)函數(shù)上指定 gpu=True。例如創(chuàng)建 tensor、讀取 CSV 文件等都適用。

import mars.tensor as mt 
 import mars.dataframe as md 
  
 a = mt.random.uniform(-1, 1, size=(1000, 1000), gpu=True) 
 df = md.read_csv('ml-20m/ratings.csv', gpu=True)

下圖是用 Mars 分別在 Scale up 和 Scale out 兩個(gè)維度上加速蒙特卡洛計(jì)算 Pi 這個(gè)任務(wù)。一般來說，我們要加速一個(gè)數(shù)據(jù)科學(xué)任務(wù)，可以有這兩種方式，Scale up 是指可以使用更好的硬件，比如用更好的 CPU、更大的內(nèi)存、使用 GPU 替代 CPU等;Scale out 就是指用更多的機(jī)器，用分布式的方式提升效率。

??

可以看到在一臺(tái) 24 核的機(jī)器上，Mars 計(jì)算需要 25.8s，而通過分布式的方式，使用 4 臺(tái) 24 核的機(jī)器的機(jī)器幾乎以線性的時(shí)間提升。而通過使用一個(gè) NVIDIA TESLA V100 顯卡，我們就能將單機(jī)的運(yùn)行時(shí)間提升到 3.98s，這已經(jīng)超越了4臺(tái) CPU 機(jī)器的性能。通過再將單卡拓展到多卡，時(shí)間進(jìn)一步降低，但這里也可以看到，時(shí)間上很難再線性擴(kuò)展了，這是因?yàn)?GPU 的運(yùn)行速度提升巨大，這個(gè)時(shí)候網(wǎng)絡(luò)、數(shù)據(jù)拷貝等的開銷就變得明顯。

性能測(cè)試

我們使用了 https://github.com/h2oai/db-benchmark 的數(shù)據(jù)集，測(cè)試了三個(gè)數(shù)據(jù)規(guī)模的 groupby 和 一個(gè)數(shù)據(jù)規(guī)模的 join。而我們主要對(duì)比了 pandas 和 DASK。DASK 和 Mars 的初衷很類似，也是試圖并行和分布式化 Python 數(shù)據(jù)科學(xué)，但它們的設(shè)計(jì)、實(shí)現(xiàn)、分布式都存在較多差異，這個(gè)后續(xù)我們?cè)僮倪M(jìn)行詳細(xì)對(duì)比。

測(cè)試機(jī)器配置是 500G 內(nèi)存、96 核、NVIDIA V100 顯卡。Mars 和 DASK 在 GPU 上都使用 RAPIDS 執(zhí)行計(jì)算。

Groupby

數(shù)據(jù)有三個(gè)規(guī)模，分別是 500M、5G 和 20G。

查詢也有三組。

查詢一：

df = read_csv('data.csv') 
 df.groupby('id1').agg({'v1': 'sum'})

查詢二：

df = read_csv('data.csv') 
 df.groupby(['id1', 'id2']).agg({'v1': 'sum'})

查詢?nèi)?/p>

df = read_csv('data.csv') 
 df.gropuby(['id6']).agg({'v1': 'sum', 'v2': 'sum', 'v3': 'sum'})

??

??

??

數(shù)據(jù)大小到 20G 時(shí)，pandas 在查詢2會(huì)內(nèi)存溢出，得不出結(jié)果。

可以看到，隨著數(shù)據(jù)增加，Mars 的性能優(yōu)勢(shì)會(huì)愈發(fā)明顯。

得益于 GPU 的計(jì)算能力，GPU 運(yùn)算性能相比于 CPU 都有數(shù)倍的提升。如果單純使用 RAPIDS cuDF，由于顯存大小的限制，數(shù)據(jù)來到 5G 都難以完成，而由于 Mars 的聲明式的特點(diǎn)，中間過程對(duì)顯存的使用大幅得到優(yōu)化，所以整組測(cè)試來到 20G 都能輕松完成。這正是 Mars + RAPIDS 所能發(fā)揮的威力。

Join

測(cè)試查詢：

x = read_csv('x.csv') 
 y = read_csv('y.csv') 
 x.merge(y, on='id1')

測(cè)試數(shù)據(jù) x 為 500M，y 包含 10 行數(shù)據(jù)。

??

總結(jié)

RAPIDS 將 Python 數(shù)據(jù)科學(xué)帶到了 GPU，極大提升了數(shù)據(jù)分析和處理的效率。Mars 的注意力更多放在并行和分布式。相信這兩者的結(jié)合，在未來會(huì)有更多的想象空間。

【本文為51CTO專欄作者“阿里巴巴官方技術(shù)”原創(chuàng)稿件，轉(zhuǎn)載請(qǐng)聯(lián)系原作者】

??戳這里，看該作者更多好文??