Mars 簡介

Mars 能利用并行和分布式技術，加速 Python 數據科學棧，包括?numpy、pandas?和?scikit-learn。同時，也能輕松與 TensorFlow、PyTorch 和 XGBoost 集成。

Mars tensor?的接口和 numpy 保持一致，但支持大規模高維數組。樣例代碼如下。

import mars.tensor as mta = mt.random.rand(10000, 50) b = mt.random.rand(50, 5000) a.dot(b).execute()

Mars DataFrame?接口和 pandas 保持一致，但可以支撐大規模數據處理和分析。樣例代碼如下。

import mars.dataframe as md ratings = md.read_csv('Downloads/ml-20m/ratings.csv') movies = md.read_csv('Downloads/ml-20m/movies.csv') movie_rating = ratings.groupby('movieId', as_index=False).agg({'rating': 'mean'}) result = movie_rating.merge(movies[['movieId', 'title']], on='movieId') result.sort_values(by='rating', ascending=False).execute()

Mars learn?保持和 scikit-learn 接口一致。樣例代碼如下。

import mars.dataframe as md from mars.learn.neighbors import NearestNeighborsdf = md.read_csv('data.csv') nn = NearestNeighbors(n_neighbors=10) nn.fit(df) neighbors = nn.kneighbors(df).fetch()

Mars learn 可以很方便地與?TensorFlow、PyTorch 和?XGBoost?集成，點擊鏈接查看文檔。

在 MaxCompute 上使用 Mars，我們提供了簡單易用的接口來拉起 Mars 集群，用戶不需要關心安裝和維護集群。同時，通過 MaxCompute 拉起的 Mars，也支持直接讀寫 MaxCompute 表。

申請試用

目前我們采用申請開通的方式，公共云用戶請提工單申請。

環境準備

要在 MaxCompute 上運行 Mars，需要有相應的運行環境。這可以分為如下幾種情況。

開箱即用的環境，如 dataworks，會包含所需要的依賴。

其他環境，需要自己安裝相關依賴。

我們分別展開。

開箱即用的環境

開箱即用的環境，如 dataworks 的?pyodps3 節點，已經包含了 PyODPS 和 Mars。

在新建的 pyodps3 節點里運行如下命令檢查版本，確保滿足要求。

from odps import __version__ as odps_version from mars import __version__ as mars_versionprint(odps_version) print(mars_version)

輸出的第一個為 PyODPS 版本，第二個為 Mars 版本。要求?PyODPS 至少是 0.9.0?。

其他環境

這個環境就要求通過 pip 安裝 PyODPS 和 Mars。Python 版本推薦使用 3.7 版本，至少需要是 3.5 版本。

通過如下命令安裝：

pip install -U pip # 可選，確保 pip 版本足夠新 pip install pyarrow==0.12.1 # 目前 pyarrow 版本固定到 0.12.1 pip install pyodps>0.9.0 # pyodps 需要至少 0.9.0 pip install pymars>=0.4.0rc1 # mars 版本需要至少是 0.4.0rc1

準備 ODPS 入口

ODPS 入口是 MaxCompute 所有操作的基礎：

• 對于開箱即用的環境，如 dataworks，我們會自動創建?o?即 ODPS 入口實例，因此可以不需要創建。
• 對于其他環境，需要通過?access_id、access_key?等參數創建，詳細參考?文檔。

基本概念

• MaxCompute 任務實例：MaxCompute 上任務以 instance 概念存在。Mars 集群也是通過一個 MaxCompute Instance 拉起。
• Logview 地址：每個 MaxCompute instance 包含一個 logview 地址來查看任務狀態。拉起 Mars 集群的 instance 也不例外。
• Mars UI: Mars 集群拉起后，會包含一個 Web UI，通過這個 Web UI，可以查看 Mars 集群、任務狀態，可以提交任務。當集群拉起后，一般來說就不需要和 MaxCompute 任務實例交互了。
• Mars session：Mars session 和具體的執行有關，一般情況下用戶不需要關心 session，因為會包含默認的 session。通過?o.create_mars_cluster?創建了 Mars 集群后，會創建默認連接到 Mars 集群的 session。
• Jupyter Notebook：Jupyter Notebook 是一個基于網頁的用于交互式計算的應用程序，可以用來開發、文檔編寫、運行代碼和展示結果。

基礎用法

創建 Mars 集群

準備好環境后，接著我們就可以拉起 Mars 集群了。

有了?o?這個對象后，拉起 Mars 集群非常簡單，只需要運行如下代碼。

from odps import options options.verbose = True # 在 dataworks pyodps3 里已經設置，所以不需要前兩行代碼 client = o.create_mars_cluster(5, 4, 16, min_worker_num=3)

這個例子里指定了 worker 數量為 5 的集群，每個 worker 是4核、16G 內存的配置，min_worker_num?指當 worker 已經起了3個后，就可以返回?client?對象了，而不用等全部 5 個 worker 都啟動再返回。Mars 集群的創建過程可能比較慢，需要耐心等待。

注意：申請的單個 worker 內存需大于 1G，CPU 核數和內存的最佳比例為 1：4，例如單 worker 4核、16G。同時，新建的 worker 個數也不要超過 30 個，否則會對鏡像服務器造成壓力，如果需要使用超過 30 個 worker，請工單申請。

這個過程中會打印 MaxCompute instance 的 logview、 Mars UI 以及 Notebook 地址。Mars UI 可以用來連接 Mars 集群，亦可以用來查看集群、任務狀態。

Mars 集群的創建就是一個 MaxCompute 任務，因此也有 instance id、logview 等 MaxCompute 通用的概念。

提交作業

Mars 集群創建的時候會設置默認 session，通過?.execute()?執行時任務會被自動提交到集群。

import mars.dataframe as md import mars.tensor as mtmd.DataFrame(mt.random.rand(10, 3)).execute() # execute 自動提交任務到創建的集群

停止并釋放集群

目前一個 Mars 集群超過3天就會被自動釋放。當 Mars 集群不再需要使用時，也可以通過調用?client.stop_server()?手動釋放：

client.stop_server()

MaxCompute 表讀寫支持

Mars 可以直讀和直寫 MaxCompute 表。

讀表

通過?o.to_mars_dataframe?來讀取 MaxCompute 表，并返回?Mars DataFrame。

In [1]: df = o.to_mars_dataframe('test_mars') In [2]: df.head(6).execute() Out[2]: col1 col2 0 0 0 1 0 1 2 0 2 3 1 0 4 1 1 5 1 2

寫表

通過?o.persist_mars_dataframe(df, 'table_name')?將 Mars DataFrame 保存成 MaxCompute 表。

In [3]: df = o.to_mars_dataframe('test_mars') In [4]: df2 = df + 1 In [5]: o.persist_mars_dataframe(df2, 'test_mars_persist') # 保存 Mars DataFrame In [6]: o.get_table('test_mars_persist').to_df().head(6) # 通過 PyODPS DataFrame 查看數據col1 col2 0 1 1 1 1 2 2 1 3 3 2 1 4 2 2 5 2 3

使用 Mars 集群自帶的 Jupyter Notebook

創建 Mars 集群會自動創建一個 Jupyter Notebook 以編寫代碼。

新建一個 Notebook 會自動設置 session，提交任務到集群。因此在這個 notebook 內也不需要顯示創建?session。

import mars.dataframe as mdmd.DataFrame(mt.random.rand(10, 3)).sum().execute() # 在 notebook 里運行，execute 自動提交到當前集群

有一點要注意：這個 notebook 不會保存你的 notebook 文件，所以要記得自行保存。

用戶也可以使用自己的 notebook 連接到集群，此時參考 使用已經創建的 Mars 集群 章節。

其他用法

使用已經創建的 Mars 集群

首先，我們可以通過 instance id 重建 Mars 集群的 client。

client = o.create_mars_cluster(instance_id=**instance-id**)

如果只是想使用 Mars，可以使用 Mars session 來連接。給定 Mars UI 的地址。則：

from mars.session import new_session new_session('**Mars UI address**').as_default() # 設置為默認 session

獲取 Mars UI 地址

Mars 集群創建的時候指定了?options.verbose=True?會打印 Mars UI 地址。

也可以通過?client.endpoint?來獲取 Mars UI。

print(client.endpoint)

獲取 Logview 地址

創建集群的時候指定了?options.verbose=True?會自動打印 logview。

也可以通過?client.get_logview_address()?獲取 logview 地址。

print(client.get_logview_address())

獲取 Jupyter Notebook 地址

Mars 集群創建的時候指定了?options.verbose=True?會打印 Jupyter Notebook 地址。

也可以通過?client.get_notebook_endpoint()?獲取 Jupyter Notebook 地址。

print(client.get_notebook_endpoint())

Mars 和 PyODPS DataFrame 對比

有同學會問，Mars 和 PyODPS DataFrame 有什么區別呢？

API

Mars DataFrame 的接口完全兼容 pandas。除了 DataFrame，Mars tensor 兼容 numpy，Mars learn 兼容 scikit-learn。

而 PyODPS 只有 DataFrame 接口，和 pandas 的接口存在著很多不同。

索引

Mars DataFrame 有 pandas 索引的概念。

In [1]: import mars.dataframe as mdIn [5]: import mars.tensor as mtIn [7]: df = md.DataFrame(mt.random.rand(10, 3), index=md.date_range('2020-5-1', periods=10)) In [9]: df.loc['2020-5'].execute() Out[9]: 0 1 2 2020-05-01 0.061912 0.507101 0.372242 2020-05-02 0.833663 0.818519 0.943887 2020-05-03 0.579214 0.573056 0.319786 2020-05-04 0.476143 0.245831 0.434038 2020-05-05 0.444866 0.465851 0.445263 2020-05-06 0.654311 0.972639 0.443985 2020-05-07 0.276574 0.096421 0.264799 2020-05-08 0.106188 0.921479 0.202131 2020-05-09 0.281736 0.465473 0.003585 2020-05-10 0.400000 0.451150 0.956905

PyODPS 里沒有索引的概念，因此跟索引有關的操作全部都不支持。

數據順序

Mars DataFrame 一旦創建，保證順序，因此一些時序操作比如?shift，以及向前向后填空值如ffill、bfill?，只有 Mars DataFrame 支持。

In [3]: df = md.DataFrame([[1, None], [None, 1]]) In [4]: df.execute() Out[4]: 0 1 0 1.0 NaN 1 NaN 1.0In [5]: df.ffill().execute() # 空值用上一行的值 Out[5]: 0 1 0 1.0 NaN 1 1.0 1.0

PyODPS 由于背后使用 MaxCompute 計算和存儲數據，而 MaxCompute 并不保證數據順序，所以這些操作在 MaxCompute 上都無法支持。

執行層

PyODPS 本身只是個客戶端，不包含任何服務端部分。PyODPS DataFrame 在真正執行時，會將計算編譯到 MaxCompute SQL 執行。因此，PyODPS DataFrame 支持的操作，取決于 MaxCompute SQL 本身。此外，每一次調用?execute?方法時，會提交一次 MaxCompute 作業，需要在集群內調度。

Mars 本身包含客戶端和分布式執行層。通過調用?o.create_mars_cluster?，會在 MaxCompute 內部拉起 Mars 集群，一旦 Mars 集群拉起，后續的交互就直接和 Mars 集群進行。計算會直接提交到這個集群，調度開銷極小。在數據規模不是特別大的時候，Mars 應更有優勢。

使用場景指引

有同學會關心，何時使用 Mars，何時使用 PyODPS DataFrame？我們分別闡述。

適合 Mars 的使用場景。

• 如果你經常使用 PyODPS DataFrame 的?to_pandas()?方法，將 PyODPS DataFrame 轉成 pandas DataFrame，推薦使用 Mars DataFrame。
• Mars DataFrame 目標是完全兼容 pandas 的接口以及行為，如果你熟悉 pandas 的接口，而不愿意學習 PyODPS DataFrame 的接口，那么使用 Mars。

• Mars DataFrame 因為兼容 pandas 的行為，因此如下的特性如果你需要用到，那么使用 Mars。

  • Mars DataFrame 包含行和列索引，如果需要使用索引，使用 Mars。
  • Mars DataFrame 創建后會保證順序，通過 iloc 等接口可以獲取某個偏移的數據。如?df.iloc[10]?可以獲取第10行數據。此外，如?df.shift()?、df.ffill()?等需要有保證順序特性的接口也在 Mars DataFrame 里得到了實現，有這方面的需求可以使用 Mars。
• Mars 還包含?Mars tensor?來并行和分布式化 Numpy，以及?Mars learn?來并行和分布式化 scikit-learn、以及支持在 Mars 集群里分布式運行 TensorFlow、PyTorch 和 XGBoost。有這方面的需求使用 Mars。
• Mars 集群一旦創建，后續不再需要通過 MaxCompute 調度，任務可以直接提交到 Mars 集群執行；此外，Mars 對于中小型任務（數據量 T 級別以下），會有較好的性能。這些情況可以使用 Mars。

適合 PyODPS DataFrame 的使用場景

• PyODPS DataFrame 會把 DataFrame 任務編譯成 MaxCompute SQL 執行，如果希望依托 MaxCompute 調度任務，使用 PyODPS DataFrame。
• PyODPS DataFrame 會編譯任務到 MaxCompute 執行，由于 MaxCompute 相當穩定，而 Mars 相對比較新，如果對穩定性有很高要求，那么使用 PyODPS DataFrame。
• 數據量特別大（T 級別以上），使用 PyODPS DataFrame。

Mars 參考文檔

• Mars 開源地址：https://github.com/mars-project/mars
• Mars 文檔：https://docs.pymars.org/zh_CN/latest/
• Mars 團隊專欄：https://zhuanlan.zhihu.com/mars-project

技術支持

技術支持請加 PyODPS 釘釘群：11701793

FAQ

Q：一個用戶創建的 Mars 集群，別人能不能用。

A：可以，參考 使用已經創建的 Mars 集群 章節。

原文鏈接
本文為云棲社區原創內容，未經允許不得轉載。

總結

以上是生活随笔為你收集整理的MaxCompute Mars 完全指南的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。