基于Spark和TensorFlow 的機器學習實踐是怎么樣的

這篇文章給大家介紹基于Spark和TensorFlow 的機器學習實踐是怎么樣的，內容非常詳細，感興趣的小伙伴們可以參考借鑒，希望對大家能有所幫助。

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是一個比較常見的推薦算法，具體場景是給電影評分，根據客戶對之前電影評分、電影類型和發布時間為用戶推薦潛在的電影。左側是一個特征工程，用戶可以使用Spark data source API讀取電影和評分信息，原生支持Spark所有操作，如join、ETL處理等；右側是TensorFlow，進行模型、優化器的選擇。

關于基于Spark和TensorFlow 的機器學習實踐是怎么樣的就分享到這里了，希望以上內容可以對大家有一定的幫助，可以學到更多知識。如果覺得文章不錯，可以把它分享出去讓更多的人看到。