Spark Shuffle 基礎(chǔ)

創(chuàng)新互聯(lián)建站專注為客戶提供全方位的互聯(lián)網(wǎng)綜合服務(wù)，包含不限于做網(wǎng)站、成都做網(wǎng)站、建平網(wǎng)絡(luò)推廣、小程序設(shè)計、建平網(wǎng)絡(luò)營銷、建平企業(yè)策劃、建平品牌公關(guān)、搜索引擎seo、人物專訪、企業(yè)宣傳片、企業(yè)代運(yùn)營等，從售前售中售后，我們都將竭誠為您服務(wù)，您的肯定，是我們最大的嘉獎；創(chuàng)新互聯(lián)建站為所有大學(xué)生創(chuàng)業(yè)者提供建平建站搭建服務(wù)，24小時服務(wù)熱線：18980820575，官方網(wǎng)址：www.cdcxhl.com

在 MapReduce 框架中，Shuffle 是連接 Map 和 Reduce 之間的橋梁，Reduce 要讀取到 Map 的輸出必須要經(jīng)過 Shuffle 這個環(huán)節(jié);而 Reduce 和 Map 過程通常不在一臺節(jié)點，這意味著 Shuffle 階段通常需要跨網(wǎng)絡(luò)以及一些磁盤的讀寫操作，因此 Shuffle 的性能高低直接影響了整個程序的性能和吞吐量。

與 MapReduce 計算框架一樣，Spark 作業(yè)也有 Shuffle 階段，通常以 Shuffle 來劃分 Stage;而 Stage 之間的數(shù)據(jù)交互是需要 Shuffle 來完成的。整個過程圖如下所示：

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

不管是 MapReduce 還是 Spark 作業(yè)，Shuffle 操作是很消耗資源的，這里的資源包括：CPU、RAM、磁盤還有網(wǎng)絡(luò);

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

而目前最新的 Spark(2.2.0) 內(nèi)置只支持一種 Shuffle 實現(xiàn)：org.apache.spark.shuffle.sort.SortShuffleManager，通過參數(shù) spark.shuffle.manager 來配置。這是標(biāo)準(zhǔn)的 Spark Shuffle 實現(xiàn)，其內(nèi)部實現(xiàn)依賴了 Netty 框架。本文并不打算詳細(xì)介紹 Spark 內(nèi)部 Shuffle 是如何實現(xiàn)的，這里我要介紹社區(qū)對 Shuffle 的改進(jìn)。

RDMA 技術(shù)

在進(jìn)行下面的介紹之前，我們先來了解一些基礎(chǔ)知識。

傳統(tǒng)的 TCP Socket 數(shù)據(jù)傳輸需要經(jīng)過很多步驟：數(shù)據(jù)先從源端應(yīng)用程序拷貝到當(dāng)前主機(jī)的 Sockets 緩存區(qū)，然后再拷貝到 TransportProtocol Driver，然后到 NIC Driver，最后 NIC 通過網(wǎng)絡(luò)將數(shù)據(jù)發(fā)送到目標(biāo)主機(jī)的 NIC，目標(biāo)主機(jī)又經(jīng)過上面步驟將數(shù)據(jù)傳輸?shù)綉?yīng)用程序，整個過程如下：

從上圖可以看出，網(wǎng)絡(luò)數(shù)據(jù)的傳輸很大一部分時間用于數(shù)據(jù)的拷貝!如果需要傳輸?shù)臄?shù)據(jù)很大，那么這個階段用的時間很可能占整個作業(yè)運(yùn)行時間的很大一部分!那么有沒有一種方法直接省掉不同層的數(shù)據(jù)拷貝，使得目標(biāo)主機(jī)直接從源端主機(jī)內(nèi)存獲取數(shù)據(jù)?還真有，這就是 RDMA 技術(shù)!

RDMA(Remote Direct Memory Access)技術(shù)全稱遠(yuǎn)程直接內(nèi)存訪問，是一種直接內(nèi)存訪問技術(shù)，它將數(shù)據(jù)直接從一臺計算機(jī)的內(nèi)存?zhèn)鬏數(shù)搅硪慌_計算機(jī)，無需雙方操作系統(tǒng)的介入。這允許高通量、低延遲的網(wǎng)絡(luò)通信，尤其適合在大規(guī)模并行計算機(jī)集群中使用(本段摘抄自 維基百科 - 遠(yuǎn)程直接內(nèi)存訪問)。RDMA 有以下幾個特點：

? Zero-copy
? 直接硬件接口(Direct hardware interface)，繞過內(nèi)核和 TCP / IP 的 IO
? 亞微秒延遲
? Flow control and reliability is offloaded in hardware

所以利用 RDMA 技術(shù)進(jìn)行數(shù)據(jù)傳輸看起來像下面一樣：

從上面看出，使用了 RDMA 技術(shù)之后，雖然源端主機(jī)和目標(biāo)主機(jī)是跨網(wǎng)絡(luò)的，但是他們之間的數(shù)據(jù)交互是直接從對方內(nèi)存獲取的，這明顯會加快整個計算過程。

SparkRDMA

好，現(xiàn)在基礎(chǔ)的知識咱們已經(jīng)獲取到了，我們正式進(jìn)入本文主題。由 Mellanox Technologies 公司開發(fā)并開源的 SparkRDMA ShuffleManager (GitHub 地址：https://github.com/Mellanox/SparkRDMA)就是采用 RDMA 技術(shù)，使得 Spark 作業(yè)在 Shuffle 數(shù)據(jù)的時候使用 RDMA 方式，而非標(biāo)準(zhǔn)的 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 就是擴(kuò)展了 Spark 的 ShuffleManager 接口，并且采用了 RDMA 技術(shù)。在測試的結(jié)果顯示，采用 RDMA 進(jìn)行 Shuffle 數(shù)據(jù)比標(biāo)準(zhǔn)的方式快 2.18 倍!

SparkRDMA 開發(fā)者們給 Spark 社區(qū)提交了一個 Issue：[SPARK-22229] SPIP: RDMA Accelerated Shuffle Engine，詳細(xì)的設(shè)計文檔：這里。不過從社區(qū)的回復(fù)來看，最少目前不會整合到 Spark 代碼中去。

安裝使用

如果你想使用 SparkRDMA，我們需要 Apache Spark 2.0.0/2.1.0/2.2.0、Java 8 以及支持 RDMA 技術(shù)的網(wǎng)絡(luò)(比如：RoCE 和 Infiniband)。

SparkRDMA 官方為不同版本的 Spark 預(yù)先編譯好相應(yīng)的 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 集群的所有節(jié)點上，其他的 jar 包只需要根據(jù)我們的 Spark 版本進(jìn)行選擇。相關(guān)的文件部署好之后，我們需要將這個 SparkRDMA 模塊加入到 Spark 的運(yùn)行環(huán)境中去，如下設(shè)置：

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 一樣。

關(guān)于配置libdisni.so

我們需要將 libdisni.so 文件分發(fā)到集群的所有節(jié)點的同一目錄下，然后配置下面的環(huán)境：

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 文件。運(yùn)行的過程中可能還需要 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"
不過如果你網(wǎng)絡(luò)不支持 RDMA 技術(shù)，那么就像我一樣會遇到下面的問題：

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 如何配置這個吧。

結(jié)語

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

如果有對大數(shù)據(jù)感興趣的小伙伴或者是從事大數(shù)據(jù)的老司機(jī)可以加群：

658558542    

歡迎大家交流分享，學(xué)習(xí)交流，共同進(jìn)步。（里面還有大量的免費(fèi)資料，幫助大家在成為大數(shù)據(jù)工程師，乃至架構(gòu)師的路上披荊斬棘?。?/p>

最后祝福所有遇到瓶頸的大數(shù)據(jù)程序員們突破自己，祝福大家在往后的工作與面試中一切順利。