2. 程式人生 > >一篇文章教你使用RDMA技術提升Spark的Shuffle效能

一篇文章教你使用RDMA技術提升Spark的Shuffle效能

阿新 • • 發佈：2019-01-14

Spark Shuffle 基礎

在 MapReduce 框架中，Shuffle 是連線 Map 和 Reduce 之間的橋樑，Reduce 要讀取到 Map 的輸出必須要經過 Shuffle 這個環節;而 Reduce 和 Map 過程通常不在一臺節點，這意味著 Shuffle 階段通常需要跨網路以及一些磁碟的讀寫操作，因此 Shuffle 的效能高低直接影響了整個程式的效能和吞吐量。

與 MapReduce 計算框架一樣，Spark 作業也有 Shuffle 階段，通常以 Shuffle 來劃分 Stage;而 Stage 之間的資料互動是需要 Shuffle 來完成的。整個過程圖如下所示：

 

 

從上面簡單的介紹可以得到以下幾個結論：

不管是 MapReduce 還是 Spark 作業，Shuffle 操作是很消耗資源的，這裡的資源包括：CPU、RAM、磁碟還有網路;

我們需要儘可能地避免 Shuffle 操作

而目前最新的 Spark(2.2.0) 內建只支援一種 Shuffle 實現：org.apache.spark.shuffle.sort.SortShuffleManager，通過引數 spark.shuffle.manager 來配置。這是標準的 Spark Shuffle 實現，其內部實現依賴了 Netty 框架。本文並不打算詳細介紹 Spark 內部 Shuffle 是如何實現的，這裡我要介紹社群對 Shuffle 的改進。

RDMA 技術

在進行下面的介紹之前，我們先來了解一些基礎知識。

傳統的 TCP Socket 資料傳輸需要經過很多步驟：資料先從源端應用程式拷貝到當前主機的 Sockets 快取區，然後再拷貝到 TransportProtocol Driver，然後到 NIC Driver，最後 NIC 通過網路將資料傳送到目標主機的 NIC，目標主機又經過上面步驟將資料傳輸到應用程式，整個過程如下：

 

 

從上圖可以看出，網路資料的傳輸很大一部分時間用於資料的拷貝!如果需要傳輸的資料很大，那麼這個階段用的時間很可能佔整個作業執行時間的很大一部分!那麼有沒有一種方法直接省掉不同層的資料拷貝，使得目標主機直接從源端主機記憶體獲取資料?還真有，這就是 RDMA 技術!

RDMA(Remote Direct Memory Access)技術全稱遠端直接記憶體訪問，是一種直接記憶體訪問技術，它將資料直接從一臺計算機的記憶體傳輸到另一臺計算機，無需雙方作業系統的介入。這允許高通量、低延遲的網路通訊，尤其適合在大規模平行計算機叢集中使用(本段摘抄自 維基百科 - 遠端直接記憶體訪問)。RDMA 有以下幾個特點：

• Zero-copy
• 直接硬體介面(Direct hardware interface)，繞過核心和 TCP / IP 的 IO
• 亞微秒延遲
• Flow control and reliability is offloaded in hardware

所以利用 RDMA 技術進行資料傳輸看起來像下面一樣：

 

 

從上面看出，使用了 RDMA 技術之後，雖然源端主機和目標主機是跨網路的，但是他們之間的資料互動是直接從對方記憶體獲取的，這明顯會加快整個計算過程。

SparkRDMA

好，現在基礎的知識咱們已經獲取到了，我們正式進入本文主題。由 Mellanox Technologies 公司開發並開源的 SparkRDMA ShuffleManager (GitHub 地址：https://github.com/Mellanox/SparkRDMA)就是採用 RDMA 技術，使得 Spark 作業在 Shuffle 資料的時候使用 RDMA 方式，而非標準的 TCP 方式。在 SparkRDMA 的官方 Wiki 裡面有如下介紹：

SparkRDMA is a high-performance, scalable and efficient ShuffleManager plugin for Apache Spark. It utilizes RDMA (Remote Direct Memory Access) technology to reduce CPU cycles needed for Shuffle data transfers. It reduces memory usage by reusing memory for transfers instead of copying data multiple times down the traditional TCP-stack.

可以看出，SparkRDMA 就是擴充套件了 Spark 的 ShuffleManager 介面，並且採用了 RDMA 技術。在測試的結果顯示，採用 RDMA 進行 Shuffle 資料比標準的方式快 2.18 倍!

 

 

SparkRDMA 開發者們給 Spark 社群提交了一個 Issue：[SPARK-22229] SPIP: RDMA Accelerated Shuffle Engine，詳細的設計文件：這裡。不過從社群的回覆來看，最少目前不會整合到 Spark 程式碼中去。

安裝使用

如果你想使用 SparkRDMA，我們需要 Apache Spark 2.0.0/2.1.0/2.2.0、Java 8 以及支援 RDMA 技術的網路(比如：RoCE 和 Infiniband)。

SparkRDMA 官方為不同版本的 Spark 預先編譯好相應的 jar 包，我們可以訪問 這裡 下載。解壓之後會得到以下四個檔案：

spark-rdma-1.0-for-spark-2.0.0-jar-with-dependencies.jar
spark-rdma-1.0-for-spark-2.1.0-jar-with-dependencies.jar
spark-rdma-1.0-for-spark-2.2.0-jar-with-dependencies.jar
libdisni.so

除了 libdisni.so 檔案一定要安裝到 Spark 叢集的所有節點上，其他的 jar 包只需要根據我們的 Spark 版本進行選擇。相關的檔案部署好之後，我們需要將這個 SparkRDMA 模組加入到 Spark 的執行環境中去，如下設定：

spark.driver.extraClassPath /path/to/SparkRDMA/spark-rdma-1.0-for-spark-2.0.0-jar-with-dependencies.jar

spark.executor.extraClassPath /path/to/SparkRDMA/spark-rdma-1.0-for-spark-2.0.0-jar-with-dependencies.jar

為了啟用 SparkRDMA Shuffle Manager 外掛，我們還需要修改 spark.shuffle.manager 的值，只需要在 $SPARK_HOME/conf/spark-defaults.conf 裡面加入以下的配合即可：

spark.shuffle.manager org.apache.spark.shuffle.rdma.RdmaShuffleManager

其他的就和正常使用 Spark 一樣。

關於配置libdisni.so

我們需要將 libdisni.so 檔案分發到叢集的所有節點的同一目錄下，然後配置下面的環境：

export JAVA_LIBRARY_PATH=$JAVA_LIBRARY_PATH:/home/iteblog/spark-2.1.0-bin/rdma/
export LD_LIBRARY_PATH=$LD_LIBRARY_PATH:/home/iteblog/spark-2.1.0-bin/rdma/
export SPARK_YARN_USER_ENV="JAVA_LIBRARY_PATH=$JAVA_LIBRARY_PATH,LD_LIBRARY_PATH=$LD_LIBRARY_PATH"

其中 /home/iteblog/spark-2.1.0-bin/rdma/ 存放了libdisni.so 檔案。執行的過程中可能還需要 libibverbs.so.1 和 librdmacm.so.1 檔案，可以通過下面命令解決

yum -y install libibverbs librdmacm

然後可以通過下面命令啟動 Spark：

bin/spark-shell --master yarn-client --driver-memory 18g --executor-memory 15g \
--queue iteblog --executor-cores 1 --num-executors 8 \
--conf "spark.yarn.dist.archives=/home/iteblog/spark-2.1.0-bin/rdma/rdma.tgz" \
--conf "spark.executor.extraLibraryPath=/home/iteblog/spark-2.1.0-bin/rdma/libdisni.so" \
--conf "spark.driver.extraLibraryPath=/home/iteblog/spark-2.1.0-bin/rdma/libdisni.so" \
--conf "spark.executor.extraClassPath=rdma.tgz/rdma/*" \
--conf "spark.driver.extraClassPath=/home/iteblog/spark-2.1.0-bin/rdma/*" \
--conf "spark.shuffle.manager=org.apache.spark.shuffle.rdma.RdmaShuffleManager"
不過如果你網路不支援 RDMA 技術，那麼就像我一樣會遇到下面的問題：

17/11/15 22:01:48 ERROR rdma.RdmaNode: Failed in RdmaNode constructor
17/11/15 22:01:48 ERROR spark.SparkContext: Error initializing SparkContext.
java.lang.reflect.InvocationTargetException
at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
at org.apache.spark.SparkEnv$.instantiateClass$1(SparkEnv.scala:265)
at org.apache.spark.SparkEnv$.create(SparkEnv.scala:323)
at org.apache.spark.SparkEnv$.createDriverEnv(SparkEnv.scala:174)
at org.apache.spark.SparkContext.createSparkEnv(SparkContext.scala:257)
at org.apache.spark.SparkContext.(SparkContext.scala:432)
at org.apache.spark.SparkContext$.getOrCreate(SparkContext.scala:2313)
at org.apache.spark.sql.SparkSession$Builder$$anonfun$6.apply(SparkSession.scala:868)
at org.apache.spark.sql.SparkSession$Builder$$anonfun$6.apply(SparkSession.scala:860)
at scala.Option.getOrElse(Option.scala:121)
at org.apache.spark.sql.SparkSession$Builder.getOrCreate(SparkSession.scala:860)
at org.apache.spark.repl.Main$.createSparkSession(Main.scala:95)
at $line3.$read$$iw$$iw.(:15)
at $line3.$read$$iw.(:42)
at $line3.$read.(:44)
at $line3.$read$.(:48)
at $line3.$read$.()
at $line3.$eval$.$print$lzycompute(:7)
at $line3.$eval$.$print(:6)
at $line3.$eval.$print()
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at scala.tools.nsc.interpreter.IMain$ReadEvalPrint.call(IMain.scala:786)
at scala.tools.nsc.interpreter.IMain$Request.loadAndRun(IMain.scala:1047)
at scala.tools.nsc.interpreter.IMain$WrappedRequest$$anonfun$loadAndRunReq$1.apply(IMain.scala:638)
at scala.tools.nsc.interpreter.IMain$WrappedRequest$$anonfun$loadAndRunReq$1.apply(IMain.scala:637)
at scala.reflect.internal.util.ScalaClassLoader$class.asContext(ScalaClassLoader.scala:31)
at scala.reflect.internal.util.AbstractFileClassLoader.asContext(AbstractFileClassLoader.scala:19)
at scala.tools.nsc.interpreter.IMain$WrappedRequest.loadAndRunReq(IMain.scala:637)
at scala.tools.nsc.interpreter.IMain.interpret(IMain.scala:569)
at scala.tools.nsc.interpreter.IMain.interpret(IMain.scala:565)
at scala.tools.nsc.interpreter.ILoop.interpretStartingWith(ILoop.scala:807)
at scala.tools.nsc.interpreter.ILoop.command(ILoop.scala:681)
at scala.tools.nsc.interpreter.ILoop.processLine(ILoop.scala:395)
at org.apache.spark.repl.SparkILoop$$anonfun$initializeSpark$1.apply$mcV$sp(SparkILoop.scala:38)
at org.apache.spark.repl.SparkILoop$$anonfun$initializeSpark$1.apply(SparkILoop.scala:37)
at org.apache.spark.repl.SparkILoop$$anonfun$initializeSpark$1.apply(SparkILoop.scala:37)
at scala.tools.nsc.interpreter.IMain.beQuietDuring(IMain.scala:214)
at org.apache.spark.repl.SparkILoop.initializeSpark(SparkILoop.scala:37)
at org.apache.spark.repl.SparkILoop.loadFiles(SparkILoop.scala:105)
at scala.tools.nsc.interpreter.ILoop$$anonfun$process$1.apply$mcZ$sp(ILoop.scala:920)
at scala.tools.nsc.interpreter.ILoop$$anonfun$process$1.apply(ILoop.scala:909)
at scala.tools.nsc.interpreter.ILoop$$anonfun$process$1.apply(ILoop.scala:909)
at scala.reflect.internal.util.ScalaClassLoader$.savingContextLoader(ScalaClassLoader.scala:97)
at scala.tools.nsc.interpreter.ILoop.process(ILoop.scala:909)
at org.apache.spark.repl.Main$.doMain(Main.scala:68)
at org.apache.spark.repl.Main$.main(Main.scala:51)
at org.apache.spark.repl.Main.main(Main.scala)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:738)
at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:187)
at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:212)
at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:126)
at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
Caused by: java.io.IOException: Unable to allocate RDMA Event Channel
at org.apache.spark.shuffle.rdma.RdmaNode.(RdmaNode.java:67)
at org.apache.spark.shuffle.rdma.RdmaShuffleManager.(RdmaShuffleManager.scala:181)
... 62 more
java.io.IOException: Unable to allocate RDMA Event Channel
at org.apache.spark.shuffle.rdma.RdmaNode.(RdmaNode.java:67)
at org.apache.spark.shuffle.rdma.RdmaShuffleManager.(RdmaShuffleManager.scala:181)
at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
at org.apache.spark.SparkEnv$.instantiateClass$1(SparkEnv.scala:265)
at org.apache.spark.SparkEnv$.create(SparkEnv.scala:323)
at org.apache.spark.SparkEnv$.createDriverEnv(SparkEnv.scala:174)
at org.apache.spark.SparkContext.createSparkEnv(SparkContext.scala:257)
at org.apache.spark.SparkContext.(SparkContext.scala:432)
at org.apache.spark.SparkContext$.getOrCreate(SparkContext.scala:2313)
at org.apache.spark.sql.SparkSession$Builder$$anonfun$6.apply(SparkSession.scala:868)
at org.apache.spark.sql.SparkSession$Builder$$anonfun$6.apply(SparkSession.scala:860)
at scala.Option.getOrElse(Option.scala:121)
at org.apache.spark.sql.SparkSession$Builder.getOrCreate(SparkSession.scala:860)
at org.apache.spark.repl.Main$.createSparkSession(Main.scala:95)
... 47 elided

這樣的話那就沒法測試了，哈哈。。如果真要使用 RDMA ，諮詢你公司的 OPS 如何配置這個吧。

結語

感謝您的觀看，如有不足之處，歡迎批評指正。

如果有對大資料感興趣的小夥伴或者是從事大資料的老司機可以加群：

658558542    

歡迎大家交流分享，學習交流，共同進步。（裡面還有大量的免費資料，幫助大家在成為大資料工程師，乃至架構師的路上披荊斬棘！）

最後祝福所有遇到瓶頸的大資料程式設計師們突破自己，祝福大家在往後的工作與面試中一切順利。


相關推薦

一篇文章教你使用RDMA技術提升Spark的Shuffle效能

Spark Shuffle 基礎 在 MapReduce 框架中，Shuffle 是連線 Map 和 Reduce 之間的橋樑，Reduce 要讀取到 Map 的輸出必須要經過 Shuffle 這個環節;而 Reduce 和 Map 過程通常不在一臺節點，這意味著 Shuffle 階段通常需要跨網路以及

一篇文章教你順利入門和開發chrome擴充套件程式

前言 關於chrome extension的開發經驗總結或說明文件等資料很多，很多人在寫，然而，我也是一員。但是，也許這篇文章，可能給你一些不一樣的感受。 前面部分大多數是一些基礎介紹，和別人的資料大同小異，但是用的是通俗的語言或者我自己理解來描述的，不是拷貝官方的描述，不然的話，你乾脆看官方文件就好啦，幹

一篇文章教你在android studio 成為Gradle高手

昨天我想弄個自動打包的玩意，結果倒騰了半天也沒倒騰好，想了想還是因為自己太懶了不願意去學習Gradle基礎知識，所以我決定今天好好把這方面的知識梳理一遍，說起來熟悉gradle開發還是很有必要的； 當然

一篇文章教你學會Git

本文轉載於掘金Ruheng，總結非常好，故不再重複造輪子。 在日常工作中，經常會用到Git操作。但是對於新人來講，剛上來對Git很陌生，操作起來也很懵逼。本篇文章主要針對剛開始接觸Git的新人，理解Git的基本原理，掌握常用的一些命令。 一、Git工作流

一篇文章教你如何用 Python 記錄日誌

轉載：http://python.jobbole.com/89007/對一名開發者來說最糟糕的情況，莫過於要弄清楚一個不熟悉的應用為何不工作。有時候，你甚至不知道系統執行，是否跟原始設計一致。線上執行的應用就是黑盒子，需要被跟蹤監控。最簡單也最重要的方式就是記錄日誌。記錄日誌

一篇文章教你如何捕獲前端錯誤

本文首發於 vivo網際網路技術 微信公眾號 https://mp.weixin.qq.com/s/E51lKQOojsvhHvACIyXwhw作者：黃文佳 常見錯誤的分類 對於使用者在訪問頁面時發生的錯誤，主要包括以下幾個型別： 1、js執行時錯誤 JavaScript程式碼在使用者瀏覽

一篇文章教你輕鬆使用fastjson

前言 只有光頭才能變強。 文字已收錄至我的GitHub精選文章，歡迎Star：https://github.com/ZhongFuCheng3y/3y JSON相信大家對他也不陌生了，前後端互動中常常就以JSON來進行資料交換。而有的時候，我們也會將JSON直接儲存在資料庫中。 可能就有人不太理解，

螞蟻技術專家：一篇文章帶你學習分布式事務

自己實現 鎖定 ons 快速 事務管理器 獲取 currency 定性 能力 小螞蟻說： 分布式事務是企業集成中的一個技術難點，也是每一個分布式系統架構中都會涉及到的一個東西，特別是在這幾年越來越火的微服務架構中，幾乎可以說是無法避免，本文就圍繞分布式事務各方面與大家進行介

一篇文章讓你瞭解大資料採集技術

大資料開啟了一個大規模生產、分享和應用資料的時代，它給技術和商業帶來了巨大的變化。麥肯錫研究表明，在醫療、零售和製造業領域，大資料每年可以提高勞動生產率0.5-1個百分點。大資料在核心領域的滲透速度有目共睹，然而調查顯示，未被使用的資訊比例高達99.4%，很大程度都是由於高價值的資訊無法獲取採集。因

無線充電技術究竟有何神祕之處？一篇文章帶你讀懂什麼是無線充電技術

無線充電技術這個概念在很早之前就已經被提出了，發展至今在電子領域已經被深入研究應用，雖然還未曾大範圍普及，但在消費電子領域的發展已經取得不錯的成績。手機廠商也紛紛在自家旗艦機上加入這一革新性的先進充電技術。那麼什麼是無線充電技術，往下看便知曉，一篇文章帶你讀懂無線充電技術。 充電運用了一

一篇文章為你解讀大數據的如今和未來

衣服 本質 海康威視 移動終端 行數據 數據分析工具 想象 下單 對手 大數據的發展前提 　 　關於大數據的概念事實上在1998年已經就有人提出了，可是到了如今才開始有所發展，這些事實上都是和當下移動互聯網的快速發展分不開的，移動互聯網的快速發展

Entity Framework學習筆記——EF簡介（一篇文章告訴你什麽是EF）

比較 編程 ast 定義 .aspx b2c 文件創建 發送 ase Entity Framework是以ADO.NET為基礎，面向數據的“實體框架”。以下簡稱EF。 它利用了抽象化數據結構的方式，將每個數據庫對象都轉換成應用程序對象 (entity)，

一篇文章讓你讀懂什麽是餐飲小程序

用戶 事情 right 連接 不用 距離 掃描二維碼 left 企業 小程序火了，餐飲+小程序也火了！ 作為連接線下消費場景和線上營銷的新工具，小程序出現後，迅速贏得了餐飲商家的青睞。與其說，小程序對餐飲行業“情有獨鐘”；不如說，兩者是“情投意合”。 20天

一篇文章讓你明白python的裝飾器

pri 9.png 不同 概念 ota 一起 數字 上下文 讀取 在看閉包問題之前先來看看關於python中作用域的問題 變量作用域 對於上述代碼中出現錯誤，肯定沒什麽疑問了，畢竟b並沒有定義和賦值，當我們把代碼更改如下後： 再看一個例子： 首先這個錯誤已經非常明顯

一篇文章帶你入門Linux——馬哥Linux基礎學習筆記

更改密碼 自帶 ctime 詳細信息 內嵌 桌面環境 地址定界 格式符 p s 1.課程體系： 中級： 初級：系統基礎 中級：系統管理、服務安全及服務管理、Shell腳本； 高級： MySQL數據庫； cache & stor

一篇文章讓你了解DNS

遞歸查詢 叠代查詢 反向解析 DNS報文格式 目錄什麽是DNSDNS的特點DNS的工作原理反向解析DNS的報文格式一、什麽是DNS DNS--Domain Name System，域名系統。 互聯網中計算機之間進行通訊，數據怎麽走，走向哪裏，這些都需要根據IP地址來

爬蟲是什麽，爬蟲能幹啥，爬蟲怎麽爬，一篇文章教會你所有

school 語句 詳細 優勢 body 谷歌 class ddd 教程 這篇文章的定位是，給有一些python基礎，但是對爬蟲一無所知的人寫的。文中只會涉及到爬蟲最核心的部分，完全避開莫名其妙的坑或概念，讓讀者覺得爬蟲是一件非常簡單的事情，而事實上爬蟲確實是一件非常簡單的

一篇文章讓你知道什麽是數據挖掘

大數據 數據挖掘 數據庫 大數據如果想要產生價值，對它的處理過程無疑是非常重要的，其中大數據分析和大數據挖掘就是最重要的兩部分。在前幾期的科普中，小編已經為大家介紹了大數據分析的相關情況，本期小編就為大家講解大數據挖掘技術，讓大家輕輕松松弄懂什麽是大數據挖掘技術。什麽是大數據挖掘？數據挖掘(Dat

一篇文章讓你豁然開朗，成為架構師你必須了解的一些疑問

有一點 同事 繼續 模式 常見 瓶頸 功能實現 很多 獨立 前言：我先介紹一下我的新同事，公司好多同事都這麽叫他James，有10年開發經驗的架構師，之前在人人網待過，之後我們也聊了很多。 在聊天過程中，我們也聊到了他為什麽離開人人網，也聊到了他在成為架構師的道路上的辛酸歷

一篇文章讓你快速入門 學懂Shell腳本

運行 turn origin 判斷語句 load for循環 網絡操作 場景 個數字 Shell腳本,就是利用Shell的命令解釋的功能，對一個純文本的文件進行解析，然後執行這些功能，也可以說Shell腳本就是一系列命令的集合。 Shell可以直接使用在win/Uni