基於 Spark 和 TensorFlow 的機器學習實踐

簡介： 大數據以及計算能力的提升，使得AI技術有了突飛猛進的發展。在大數據和AI技術的熱潮下，在2019杭州雲棲大會機器學習技術專場，阿里雲高級技術專家吳威和阿里雲技術專家江宇向大家分享了EMR E-Learning平臺和平臺上新開發的核心特性TensorFlow on Spark。

EMR E-Learning平臺

EMR E-Learning平臺基於的是大數據和AI技術，通過算法基於歷史數據來構建機器學習模型，從而進行訓練與預測。目前機器學習被廣泛應用到很多領域，如人臉識別、自然語言處理、推薦系統、計算機視覺等。近年來，大數據以及計算能力的提升，使得AI技術有了突飛猛進的發展。

機器學習中重要的三要素是算法、數據和算力。而EMR本身是一個大數據平臺，平臺之上擁有多種數據，比如傳統的數據倉庫數據、圖像數據；EMR有很強的調度能力，可以很好地吊調度GPU和CPU資源；其結合機器學習算法，就可以成為一個比較好的AI平臺。

典型的AI開發流程如下圖所示：首先是數據收集，手機、路由器或者日誌數據進入大數據框架Data Lake；然後是數據處理，收集到的數據需要通過傳統的大數據ETL或特徵工程進行處理；其次是模型訓練，經過特徵工程或ETL處理後的數據會進行模型的訓練；最後對訓練模型進行評估和部署；模型預測的結果會再輸入到大數據平臺進行處理分析，整個過程循環往復。

下圖展示了AI開發的流程，左側是單機或者集群，主要進行AI訓練和評估，包含數據存儲；右側是大數據存儲，主要進行大數據處理，如特徵工程等，同時可以利用左側傳輸的機器學習模型進行預測。

AI開發的現狀主要有以下兩點：

• 兩套集群運維複雜：從圖中可以看出，AI開發涉及的兩套集群是分離的，需要單獨維護，運維成本複雜，容易出錯。

• 訓練效率較低：左右兩側集群需要大量數據傳輸和模型傳輸，帶來較高的端到端訓練的延遲。

EMR作為統一的大數據平臺，包含了很多特性。最底層基礎設施層，其支持GPU和CPU機器；數據存儲層包括HDFS和阿里雲OSS；數據接入層包括Kafka和Flume；資源調度層計算引擎包括 YARN、K8S和Zookeeper；計算引擎最核心的是E-learning平臺，基於目前比較火的開源系統Spark，這裡的Spark用的是jindo Spark，是EMR團隊基於Spark改造和優化而推出的適用於AI場景下的版本，除此之外，還有PAI TensorFlow on Spark；最後是計算分析層，提供了數據分析、特徵工程、AI訓練以及Notebook的功能，方便用戶來使用。

EMR平臺的特性主要有以下幾點：

• 統一的資源管理與調度：支持CPU、Mem和GPU的細粒度的資源調度和分配，支持YARN和K8S的資源調度框架；

• 多種框架支持：包括TensorFlow、MXNet和Caffe等；

• Spark通用的數據處理框架：提供Data Source API來方便各類數據源的讀取，MLlib pipeline廣泛用於特徵工程；

• Spark+深度學習框架：Spark和深度學習框架的集成支持，包括高效的Spark和TensorFlow之間的數據傳輸，Spark資源調度模型支持分佈式深度學習訓練；

• 資源監控與報警：EMR APM系統提供完善的應用程序和集群監控多種報警方式；

• 易用性：Jupyter notebook以及Python多環境部署支持，端到端機器學習訓練流程等。

EMR E-Learning集成了PAI TensorFlow，其支持對深度學習的優化和對大規模稀疏場景的優化。

TensorFlow on Spark

經過市場調研發現，大多數的客戶在深度學習之前的數據ETL和特徵工程階段使用的都是開源計算框架Spark，之後的階段廣泛使用的是TensorFlow,因此就有了將TensorFlow和Spark有機結合的目標。TensorFlow on Spark主要包含了下圖中的六個具體設計目標。

TensorFlow on Spark從最底層來講實際上是PySpark應用框架級別的封裝。框架中實現的主要功能包括：首先調度用戶特徵工程任務，然後再調度深度學習TensorFlow任務，除此之外還需要將特徵工程的數據高效快速地傳輸給底層的PAI TensorFlow Runtime進行深度學習和機器學習的訓練。由於Spark目前不支資源的異構調度，假如客戶運行的是分佈式TensorFlow， 就需要同時運行兩個任務（Ps任務和Worker任務），根據客戶需求的資源來產生不同的Spark executor，Ps任務和Worker任務通過Zookeeper來進行服務註冊。框架啟動後會將用戶寫的特徵工程任務調度到executor中執行，執行後框架會將數據傳輸給底層的PAI TensorFlow Runtime進行訓練，訓練結束後會將數據保存到Data Lake中，方便後期的模型發佈。

在機器學習和深度學習中，數據交互是可以提升效率的點，因此在數據交互部分，TensorFlow on Spark做了一系列優化。具體來講採用了Apache Arrow進行高速數據傳輸，將訓練數據直接餵給API TensorFlow Runtime,從而加速整個訓練流程。

TensorFlow on Spark的容錯機制如下圖所示：最底層依賴TensorFlow的Checkpoints機制，用戶需要定時的將訓練模型Chenpoint到Data Lake中。當重新啟動一個TensorFlow的時候，會讀取最近的Checkpoint進行訓練。容錯機制會根據模式不同有不同的處理方式，針對分佈式任務，會啟動Ps和Worker任務，兩個任務直接存在daemon進程，監控對應任務運行情況；對於MPI任務，通過Spark Barrier Execution機制進行容錯，如果一個task失敗，會標記失敗並重啟所有task，重新配置所有環境變量；TF任務負責讀取最近的Checkpoint。

TensorFlow on Spark的功能和易用性主要體現在以下幾點：

• 部署環境多樣：支持指定conda，打包python運行時virtual env 支持指定docker

• TensorFlow 架構支持：支持分佈式TensorFlow原生PS架構和分佈式Horovod MPI架構

• TensorFlow API支持：支持分佈式TensorFlow Estimator高階API和分佈式TensorFlow Session 低階API

• 快速支持各種框架接入：可以根據客戶需求加入新的AI框架，如MXNet

EMR客戶有很多來自於互聯網公司，廣告和推送的業務場景比較常見，下圖是一個比較典型的廣告推送業務場景。整個流程是EMR客戶通過Kafka將日誌數據實時推送到Data Lake中，TensorFlow on Spark負責的是上半部分流程，其中可以通過Spark的工具如SparkSQL、MLlib等對實時數據和離線數據進行ETL和特徵工程，數據訓練好之後可以通過TensorFlow 框架高效地餵給PAI TensorFlow Runtime進行大規模訓練和優化，然後將模型存儲到Data Lake中。

在API層面，TensorFlow on Spark提供了一個基類，該基類中包含了三個方法需要用戶去實現：pre_train、shutdown和train。pre_train是用戶需要做的數據讀取、ETL和特徵工程等任務，返回的是Spark的DataFrame對象；shutdown方法實現用戶長連接資源的釋放；train方法是用戶之前在TensorFlow中實現的代碼，如模型、優化器、優化算子的選擇。最後通過pl_submit命令來提交TensorFlow on Spark的任務。

下圖是推薦系統FM的樣例，FM是一個比較常見的推薦算法，具體場景是給電影評分，根據客戶對之前電影評分、電影類型和發佈時間為用戶推薦潛在的電影。左側是一個特徵工程，用戶可以使用Spark data source API讀取電影和評分信息，原生支持Spark所有操作，如join、ETL處理等；右側是TensorFlow，進行模型、優化器的選擇。目前整個系統的代碼已經開源到Github。

最後總結一下，EMR E-Learning 平臺將大數據處理、深度學習、機器學習、數據湖、GPUs功能特性緊密的結合，提供一站式大數據與機器學習平臺；TensorFlow on Spark 提供了高效的數據交互流程以及完備的機器學習訓練流程，將Spark與TensorFlow結合，藉助 PAI TensorFlow，助力用戶加速訓練；目前E-Learning平臺在公有云服務不同的客戶，成功案例，CPU集群規模超過1000，GPU集群規模超過100。

點擊“瞭解更多”觀看視頻&PPT資料