spark杰普軟件科技有限公司

1、杰普軟件科技公司總部：上海市閘北區(qū)萬榮路1188弄龍軟軟件園區(qū)A棟206室：（021）56778147：200436昆山實(shí)訓(xùn)基地：昆山市巴城學(xué)院路828號(hào)昆山浦東軟件園北樓4-5層：（0512）50190290-8000：215311Brighten Your Way And Raise You Up.電郵：training主頁：1Briup Training What is Spark Spark簡介( Spark4個(gè)特點(diǎn)與發(fā)展) BDAS系統(tǒng) Why Spark Spark與MapReduce的差異 Spark的適用場景 Spark How Spark案例安裝Spark Spark常用命令

2、Spark架構(gòu)組件和工作機(jī)制Spark運(yùn)行原理Spark-Core內(nèi)容內(nèi)容Spark-SqlSpark-Streaming內(nèi)容2杰普軟件科技公司總部：上海市閘北區(qū)萬榮路1188弄龍軟軟件園區(qū)A棟206室：（021）56778147：200436昆山實(shí)訓(xùn)基地：昆山市巴城學(xué)院路828號(hào)昆山浦東軟件園北樓4-5層：（0512）50190290-8000：215311Brighten Your Way And Raise You Up.電郵：training主頁：3Briup Training Spark是加州大學(xué)伯克利分校AMP開發(fā)通用內(nèi)存并行計(jì)算框架。 Apache Spark is a fast

3、 and general engine for large-scale data processing. Spark是基于內(nèi)存計(jì)算的大數(shù)據(jù)分布式計(jì)算框架。Spark基于內(nèi)存計(jì)算，提用戶將Spark部署在大供可交互的方式，提供近實(shí)時(shí)的處理方式，量廉價(jià)硬件之上，形成集群，提高處理速度，同時(shí)也保證了集群高容錯(cuò)性和高可伸縮性。4Briup Training 2009年伯克利分校開始使用Scala語言編寫Spark源碼，2010年開放源代碼，2013年6月進(jìn)入Apache孵化項(xiàng)目，2014年2月成為Apache頂級(jí)項(xiàng)目，2014年5月底發(fā)布Spark1.0.0,到目前為止版本更新為Spark2.4.3。

4、5Briup Training 演進(jìn)時(shí)間表：2009年由 Berkeleys AMPLab 開始編寫最初的源代碼2010年開放源代碼2013年6月進(jìn)入Apache孵化器項(xiàng)目2014年2月成為Apache的頂級(jí)項(xiàng)目（82014年5月底Spark1.0.0發(fā)布2014年9月Spark1.1.0發(fā)布.2016年3月Spark1.6.1發(fā)布2016年7月Spark2.0發(fā)布2016年10月Spark2.0.1發(fā)布2016年11月Spark1.6.3發(fā)布2016年11月Spark2.0.2發(fā)布.2017年10月Spark2.1.2發(fā)布2017年12月Spark2.2.1發(fā)布2018年2月Spark2.3

5、.0發(fā)布時(shí)間）6Briup Training 在Spark官網(wǎng)上介紹，它具有運(yùn)行速度快、易用性好、通用性強(qiáng)和隨處運(yùn)行等特點(diǎn)。7Briup Training 運(yùn)行速度快 Spark擁有DAG執(zhí)行引擎，支持在內(nèi)存中對(duì)數(shù)據(jù)進(jìn)行迭代計(jì)算。提供的數(shù)據(jù)，速度是Hadoop MapReduce的10倍以上，如果數(shù)據(jù)表明，如果數(shù)據(jù)由磁盤，速度可以高達(dá)100多倍。從內(nèi)存中8Briup Training 易用性 支持4種語言的API：scala、java、python、R。特別是Scala是一種高效、可拓展的語言，能夠用簡潔的代碼處理較為復(fù)雜的處理工作。 val text_file=sc.textFile(“ h

6、dfs:/.” ) text_file.flatMap(_.split(“ text_file.flatMap(_.split(“” ).map(x = (x,1).reduceByKey(_+_)” ).countByValue();9Briup Training scala val textFile = spark.read.textFile(“ README.md) val wordCounts = textFile.flatMap(line = line.split( ).groupByKey(identity).count()1Briup Training 通用性強(qiáng) Spark圈即B

7、DAS（伯克利數(shù)據(jù)分析棧）包含了Spark Core、Spark SQL、Spark Streaming、MLLib和GraphX等組件，這些組件分別處理Spark Core提供內(nèi)存計(jì)算框架、SparkStreaming的實(shí)時(shí)處理應(yīng)用、Spark SQL的即席、MLlib或MLbase的學(xué)習(xí)和GraphX的圖處理，它們都是由AMP提供，能夠無縫的集成并提供一站式解決平臺(tái)。1Briup Training 隨處運(yùn)行 Spark具有很強(qiáng)的適應(yīng)性，能夠HDFS、HBase、Cassandra S3和Techyon為持久層讀寫原生數(shù)據(jù)，能夠以Mesos、YARN和自身攜帶的Standalone作為管理器

8、調(diào)度job，來完成Spark應(yīng)用程序的計(jì)算。12Briup Training Spark圈也稱為BDAS（伯克利數(shù)據(jù)分析棧），是伯克利APMLab打造的，力圖在算法（Algorithms）、（Machines）、人（People）之間通過大規(guī)模集成來展現(xiàn)大數(shù)據(jù)應(yīng)用的一個(gè)平臺(tái)。伯克利AMPLab運(yùn)用大數(shù)據(jù)、云計(jì)算、通信等各種以及各種靈活的技術(shù)方案，對(duì)海量不透明的數(shù)據(jù)進(jìn)行甄別并轉(zhuǎn)化為有用的信息，以供人們更好的理解世界。該圈已經(jīng)涉及到學(xué)習(xí)、數(shù)據(jù)挖掘、數(shù)據(jù)庫、信息檢索、自然語言處理和語音識(shí)別等多個(gè)領(lǐng)域。13Briup Training Spark圈以Spark Core為，從HDFS、HBase和Am

9、azon S3等持久層數(shù)據(jù)，以MESOS、YARN和自身攜帶的Standalone為管理器調(diào)度Job完成Spark應(yīng)用程序的計(jì)算。 這些應(yīng)用程序可以來自于不同的組件，如SparkShell/Spark Submit的批處理、Spark Streaming的實(shí)時(shí)處理應(yīng)用、Spark SQL、BlinkDB的權(quán)衡、MLlib/MLbase的學(xué)習(xí)、GraphX的圖處的即席理和SparkR的數(shù)學(xué)計(jì)算等等。14Briup Training15杰普軟件科技公司總部：上海市閘北區(qū)萬榮路1188弄龍軟軟件園區(qū)A棟206室：（021）56778147：200436昆山實(shí)訓(xùn)基地：昆山市巴城學(xué)院路828號(hào)昆山浦東軟

10、件園北樓4-5層：（0512）50190290-8000：215311Brighten Your Way And Raise You Up.電郵：training主頁：16Briup Training Spark是在借鑒了MapReduce之上發(fā)展而來的，繼承了其分布式并行計(jì)算的優(yōu)點(diǎn)并改進(jìn)了MapReduce明顯的缺陷，具體如下：基于內(nèi)存計(jì)算容錯(cuò)性高通用性好17Briup Training 首先，Spar k把中間數(shù)據(jù)放到內(nèi)存中，迭代運(yùn)算效率高。MapReduce中計(jì)算結(jié)果需要落地，保存到磁盤上，這樣勢(shì)必會(huì)影響整體速度，而Spar k支持DAG圖的分布式并行計(jì)算的編程框架，減少了迭代過程中數(shù)據(jù)

11、的落地，提高了處理效率。 其次，Spar k容錯(cuò)性高。Spar k引進(jìn)了彈性分布式數(shù)據(jù)集RDD (Resilient Distr ibuted Dataset) 的抽象，它是分布在一組節(jié)點(diǎn)中的只讀對(duì)象集合，這些集合是彈性的，如果數(shù)據(jù)集一部分丟失，則可以根據(jù)“血統(tǒng)”（即充許基于數(shù)據(jù)衍生過程）對(duì)它們進(jìn)行重建。另 外在RDD計(jì)算時(shí)可以通過CheckPoint來實(shí)現(xiàn)容錯(cuò)，而CheckPoint有兩種方式：CheckPoint Data，和Logging The Updates，用戶可以采用哪種方式來實(shí)現(xiàn)容錯(cuò)。 最后，Spar k更加通用。不像Hadoop只提供了Map和Reduce兩種操作，Spar

12、k提供的數(shù)據(jù)集操作類型有很多種，大致分為：Transfor mations和Actions兩大類。Transfor mations包括Map、Filter、FlatMap、Sample、Gr oupByKey、ReduceByKey、Union、J oin、Cogroup、MapValues、Sor t和PartionBy等多種操作類型，同時(shí)還提供Count, Actions包括Collect、Reduce、Lookup和Save等操作。另外各個(gè)處理節(jié)點(diǎn)之間的通信模型不再像Hadoop只有Shuffle一種模式，用戶可以命名、物化，、分區(qū)等。中間結(jié)果的18Briup Training 目前大數(shù)

13、據(jù)處理場景有以下幾個(gè)類型： 1. 復(fù)雜的批量處理（Batch Data Processing），偏重點(diǎn)在于處理海量數(shù)據(jù)的能力， 至于處理速度可忍受，通常的時(shí)間可能是在數(shù)十分鐘到數(shù)小時(shí)； 2. 基于歷史數(shù)據(jù)的交互式分鐘之間（Interactive Query），通常的時(shí)間在數(shù)十秒到數(shù)十 3. 基于實(shí)時(shí)數(shù)據(jù)流的數(shù)據(jù)處理（Streaming Data Processing），通常在數(shù)百毫秒到數(shù)秒之間 目前對(duì)以上三種場景需求都有比較成處理框架，第一種情況可以用Hadoop的MapReduce來進(jìn)行批量海量數(shù)據(jù)處理，第二種情況可以Impala進(jìn)行，對(duì)于第三中情況可以用Storm分布式處理框架處理實(shí)時(shí)流式

14、數(shù)據(jù)。交互式，各自一套維護(hù)成本比較高，而Spark的出現(xiàn)能夠一站以上三者都是比較式平臺(tái)滿意以上需求。19Briup Training Spark是基于內(nèi)存的迭代計(jì)算框架，適用于需要多次操作特定數(shù)據(jù)集的應(yīng)用場合。需要反復(fù)操作的次數(shù)越多，所需的數(shù)據(jù)量越大，受益越大，數(shù)據(jù)量小但是計(jì)算密集度較大的場合，受益就相對(duì)較小 由于RDD的特性，Spark不適用那種異步細(xì)粒度更新狀態(tài)的應(yīng)用，例如web服或者是增量的web爬蟲和索引。就是對(duì)于那種增量修改的應(yīng)用模型不務(wù)的適合 數(shù)據(jù)量不是特別大，但是要求實(shí)時(shí)統(tǒng)計(jì)分析需求20Briup TrainingYahoo將Spark用在Audience Expansion中的

15、應(yīng)用。Audience Expansion是廣告中尋找目標(biāo)用戶的法：首先者提供一些了并且的樣本客戶，據(jù)此進(jìn)行學(xué)習(xí)，尋找可能轉(zhuǎn)化的用戶，對(duì)他們定向。Yahoo采用的算法是logistic regression。同時(shí)由于有些SQL負(fù)載需要更高的服務(wù)質(zhì)量，又加入了專門跑Spark的大內(nèi)存集群，用于取代商業(yè)BI/OLAP工具，承擔(dān)報(bào)表/圖表和交互式/即席，同時(shí)與桌面BI工具對(duì)接。目前在Yahoo部署的Spark集群有112臺(tái)節(jié)點(diǎn)，9.2TB內(nèi)存。業(yè)務(wù)，最初使用Mahout或者寫的MR來解決復(fù)雜的阿里搜索和學(xué)習(xí)，導(dǎo)致效率低而且代碼不易維護(hù)。淘寶技術(shù)團(tuán)隊(duì)使用了Spark來解決多次迭學(xué)習(xí)算法、高計(jì)算復(fù)雜度的

16、算法等。將Spark運(yùn)用于淘寶的推薦相關(guān)代的算法上,同時(shí)還利用Graphx解決了許多生產(chǎn)問題，包括以下計(jì)算場景：基于度分布的中樞節(jié)點(diǎn)發(fā)現(xiàn)、基于最大連通圖的社區(qū)發(fā)現(xiàn)、基于三角形計(jì)數(shù)的關(guān)系衡量、基于隨機(jī)游走的用戶屬性等。21杰普軟件科技公司總部：上海市閘北區(qū)萬榮路1188弄龍軟軟件園區(qū)A棟206室：（021）56778147：200436昆山實(shí)訓(xùn)基地：昆山市巴城學(xué)院路828號(hào)昆山浦東軟件園北樓4-5層：（0512）50190290-8000：215311Brighten Your Way And Raise You Up.電郵：training主頁：22Briup Training Spark運(yùn)行

17、模式 Local本地模式(多用于測試) Standalone模式 Yarn集群模式 Mesos集群模式 Kubernetes集群模式 Spark環(huán)境搭建 Spark常用命令 spark-shell spark-submit spark-sql Spark基礎(chǔ)案例23Briup Training 1.解壓tar -xvf spark.2.2.1-bin-hadoop2.7.tar 2.mv spark.2.2.1-bin-hadoop2.7 /opt/software/ 3.修改環(huán)境變量： vi .bashrc export SPARK_HOME=/usr/spark/spark-2.2.1-b

18、in-hadoop2.7 exportPATH=$PATH:$SPARK_HOME/bin:$PATH:$SPARK_HOME/sbin 4.source .bashrc 5.(可選)修改下log4j的級(jí)別 $SPARK_HOME/conf/pertie 6.測試spark是否安裝 spark-shell：24Briup Training 測試環(huán)境run-example SparkPi 10 測試spark-submit命令 $Spark_home/spark-submit-class org.apache.spark.examples.SparkPi examples/ja

19、rs/spark-examples_2.11-2.2.1.jar1025Briup Training 1.修改Spark配置文件 (路徑為$SPARK_HOME/conf/下)#cp cpslaves.template和 spark-env.sh.template各一份spark-env.sh.template spark-env.shslaves.template slaves 2.vi slaves，此文件是指定子節(jié)點(diǎn)的主機(jī),直接添加從節(jié)點(diǎn)主機(jī)名即可(前提是在/etc/hosts里面配置了).例如：mm0 mm1mm226Briup Training3.在spark-env.sh末端添加如

20、下幾行：#必須配置：#主節(jié)點(diǎn)的IP地址export SPARK_MASTER_IP=00#主節(jié)點(diǎn)的端export SPARK_MASTER_PORT=7077#指定Spark用來混洗數(shù)據(jù)的本地路徑export SPARK_LOCAL_DIRS=/data/spark/dirs,/home/briup/spark/dirs（一定要注意這個(gè)混洗數(shù)據(jù)的路徑的權(quán)限）27Briup Training#可選配置：#Worker的WebUI端export SPARK_WORKER_WEBUI_PORT=8081#主節(jié)點(diǎn)的WEBUI端export SPARK_MASTER_WEBUI_P

21、ORT=8099#每個(gè)Worker使用的CPU核數(shù),默認(rèn)1個(gè)export SPARK_WORKER_CORES=2#每個(gè)Slave中啟動(dòng)幾個(gè)Worker實(shí)例,默認(rèn)1個(gè)export SPARK_WORKER_INSTANCES=2#每個(gè)Worker使用多大的內(nèi)存,默認(rèn)1g export SPARK_WORKER_MEMORY=2g #驅(qū)動(dòng)器節(jié)點(diǎn)Driver使用的內(nèi)存export SPARK_DRIVER_MEMORY=2g28Briup Training 4.啟動(dòng)start-all.sh命令或者是start-master.sh/start-slave.sh spark:/host:7077 5.

22、測試是否上輸入jps頁面，在集群的所有 6.打開localhost:8080查看集群29Briup Training 注意事項(xiàng)： 1.默認(rèn)情況下，集群和Mesos提交的Application會(huì)使用所有集群中剩余的CPU Core。如果要限定每個(gè)提交的Spark Application的使用的CPU核的數(shù)目,可以采用下邊的辦法： 在sparkconf對(duì)象中設(shè)置spark.cores.max 核數(shù)。例如:val conf = new SparkConf() .setMaster(.).setAppName(.) .set( spark.cores.max , 10 ) 在群集上更改應(yīng)用程序設(shè)置應(yīng)該

23、使用spark.deploy.defaultCores。例如：在conf/ spark-env.sh： export SPARK_MASTER_OPTS=“-Dspark.deploy.defaultCores= 在spark2.2.1版本之后，需要在使用spark-submit或者spark-shell命令時(shí)，通過 total-executor-cores 2來進(jìn)行設(shè)置。 2.必須確保executor_memory的值= worker_memory的值30Briup Training基于文件系統(tǒng)的單點(diǎn)恢復(fù)(Single-Node Recovery with Local File System

24、)export SPARK_DAEMON_JAVA_OPTS=-Dspark.deploy.recoveryMode=FILESYSTEM -Dspark.deploy.recoveryDirectory=/opt/software/spark- 1.6.1-bin-hadoop2.6/recovery基于zookeeper的Standby Masters(Standby Masters with ZooKeeper) export SPARK_DAEMON_JAVA_OPTS=-Dspark.deploy.recoveryMode=ZOOKEEPER -Dspark.deploy.zooke

25、eper.url=17:2181, 18:2181,19:2181 -Dspark.deploy.zookeeper.dir=/spark:用于指定Master的HA，依賴于zookeeper集群;31Briup Training 測試spark-submit命令 $Spark_home/spark-submit-master spark:/host:7077-class org.apache.spark.examples.SparkPi examples/jars/spark-examples_2.11-2.2.1.jar10

26、32Briup Training 1.只需要修改spark-env.sh環(huán)境配置即可： 2.添加如下配置：export HADOOP_CONF_DIR =/opt/hadoop/etc/hadoopexport SPARK_LOCAL_IP=你的IP 優(yōu)化配置#將spark的類庫jar包上傳到hdfs上，省的每次都上傳Spark1.*版本export SPARK_JAR=hdfs:/:9000/user/spark/share/lib/spark- assembly.jarSpark2.*版本spark.yarn.jars或者spark.yarn.archivehdfs:/:9000/spar

27、k_lib/jars/*hdfs:/:9000/spark_lib/jars33Briup Training 注意事項(xiàng)： 在yarn集群上部署spark,即添加HADOOP_CONF_DIR到spark-env.sh中, 不論要運(yùn)行本地spark-shell還是yarn集群spark-shell,必須先啟動(dòng)yarn集群。34Briup Training 1.將hive-site.xml添加到$SPARK_HOME/conf/目錄下； 2.集群運(yùn)行模式下，需要將Hive的元數(shù)據(jù)的連接jar包配置到執(zhí)行器節(jié)點(diǎn)； spark.executor.extraClassPath sql-connector

28、-java-5.1.22.jar 3.設(shè)置warehouse$HIVE_HOME/lib/my spark.sql.warehouse.dir hdfs:/computer1.cloud.briu :9000/user/hive/warehouse/ 4.測試 在本地測試只能通過 -driver-class-path 選項(xiàng)將Hive的元數(shù)據(jù)的連接jar包添加到Driver中。 spark-sql-driver-class-path $HIVE_HOME/lib/mysql-connector-java-5.1.22.jar35Briup Trainingspark-shellspark-sub

29、mit spark-sql36Briup Training 格式：./bin/spark-shell optionsoptions可選參數(shù) 可通過spark-shell-help查看eg:spark-shell -master local3 -name mySparkShell37Briup Training 格式：Usage: spark-submit options app argumentsUsage: spark-submit -kill submission ID -master spark:/. Usage: spark-submit -status submission ID -

30、master spark:/.Usage: spark-submit run-example options example-class exampleargsoptions可選參數(shù) 可通過spark-submit -help查看表示包含應(yīng)用的jar包app arguments傳給應(yīng)用的選項(xiàng)38Briup Training spark-submit的常見標(biāo)記-master 集群URL-deploy-mode 工作模式-class 應(yīng)用程序的主類-name 應(yīng)用的名稱-jars 額外依賴的第jar包-files 需要分發(fā)到各節(jié)點(diǎn)的數(shù)據(jù)文件-executor-memory 執(zhí)行器內(nèi)存大小-driv

31、er-memory 驅(qū)動(dòng)器內(nèi)存大小-conf prop=value標(biāo)記設(shè)置SparkConf配置選項(xiàng)-properties-file 指定一個(gè)包含鍵值對(duì)的屬性文件. eg:spark-submit-master local -class com.briup.scala.First*.jar39Briup Training 格式：./bin/spark-sql options cli optionoptions可選參數(shù) 可通過spark-sql-help查看cli options可選參數(shù)可通過spark-sql-help查看eg:spark-sql-master local40Briup Tra

32、ining Scala行數(shù)統(tǒng)計(jì)val lines=sc.textFile(“ readme.md” ) lines.count()lines.first()思考：1. 哪來的sc?2. sc是什么?41Briup Training 驅(qū)動(dòng)器程序(Driver Program) 包含應(yīng)用的main函數(shù)，并定義了集群上的分布式數(shù)據(jù)集，并對(duì)分布式數(shù)據(jù)集應(yīng) 用相關(guān)操作。 執(zhí)行器(executor) sc(SparkContext)42Briup Training構(gòu)建SparkContext對(duì)象 在一個(gè)的項(xiàng)目中，需要 pom文件中添加spark-core依賴org.apache.sparkspark-co

33、re_43Briup Training 代碼編寫結(jié)構(gòu)如下：SparkConfval conf=newSparkConfval sc=newSparkContext(conf)SparkContextval rdd:RDDT=sc.方法(參數(shù))RDDsc.stop()關(guān)閉SparkContext對(duì)象44Briup Training45Briup Training46杰普軟件科技公司總部：上海市閘北區(qū)萬榮路1188弄龍軟軟件園區(qū)A棟206室：（021）56778147：200436昆山實(shí)訓(xùn)基地：昆山市巴城學(xué)院路828號(hào)昆山浦東軟件園北樓4-5層：（0512）50190290-80

34、00：215311Brighten Your Way And Raise You Up.電郵：training主頁：47Briup TrainingBriup Training Spark架構(gòu)圖49Briup Training50Briup Training Executor:執(zhí)行器，Application運(yùn)行在worker節(jié)點(diǎn)上的一個(gè)進(jìn)程，負(fù)責(zé)啟動(dòng)線程池運(yùn)行任務(wù)(Task),并且負(fù)責(zé)將數(shù)據(jù)存在內(nèi)存或磁盤上。每個(gè)Application擁有executors。在Spark on Yarn 模式下， 其進(jìn)程名稱為的一組CoarseGrainedExecutorBackend， 類似于 Hadoop

35、 MapReduce 中的 YarnChild。一個(gè) CoarseGrainedExecutorBackend 進(jìn)程有且僅有一個(gè)executor 對(duì)象，它負(fù)責(zé)將 Task 包裝成 taskRunner，并從線程池中抽取出一個(gè)空閑線程運(yùn)行 Task。每個(gè)CoarseGrainedExecutorBackend 能并行運(yùn)行 Task 的數(shù)量就取決于分配給它的CPU 的個(gè)數(shù)了；51Briup Training 彈性分布式數(shù)據(jù)集(RDD)Resillient Distributed Dataset，Spark 的基本計(jì)算單元，可以通過一系列算子進(jìn)行操作（主要有 Transformation 和 Act

36、ion 操作,一組RDD形成執(zhí)行的有向無環(huán)圖RDD Graph。 有向無環(huán)圖(DAG)DirectedSched Acycle graph，反應(yīng) RDD 之間的依賴關(guān)系； RDD DAG當(dāng)RDD遇到Action算子，將之前的所有算子形成一個(gè)有向無環(huán)圖。在Spark中轉(zhuǎn)化為Job，提交到集群進(jìn)行執(zhí)行。一個(gè)App中可以包含多個(gè)Job。 有向無環(huán)圖調(diào)度器(DAG Scheduler)根據(jù) Job 構(gòu)建基于 Stage 的 DAG，并提交 Stage給 TaskScheduler; 任務(wù)調(diào)度器(Task Scheduler):將Task分發(fā)給Executor執(zhí)行。(將Taskset提交給worker(

37、集群)運(yùn)行并回報(bào)結(jié)果); SparkEnv線程級(jí)別的上下文，運(yùn)行時(shí)的重要組件的。52Briup Training53Briup Training54Briup Training Spark運(yùn)行基本流程55Briup Training56Briup Training Spark on Standalone運(yùn)行過程57Briup Training58Briup Training59Briup Training60Briup Training 在YARN中，每個(gè)Application實(shí)例都有一個(gè)ApplicationMaster進(jìn)程，它是Application啟動(dòng)的第一個(gè)容器。它負(fù)責(zé)和Resourc

38、eManager打交道并請(qǐng)求之后告訴NodeManager為其啟動(dòng)Container。從獲取次的含義講YARN-Cluster和YARN-Client模式的區(qū)別其實(shí)就是ApplicationMaster進(jìn)程的區(qū)別。 YARN-Cluster模式下，Driver運(yùn)行在AM(Application Master)中，它負(fù)責(zé)向YARN申請(qǐng)，并監(jiān)督作業(yè)的運(yùn)行狀況。當(dāng)用戶提交了作業(yè)之后，就可以關(guān)掉Client，作業(yè)會(huì)繼續(xù)在YARN上運(yùn)行，因而YARN-Cluster模式不適合運(yùn)行交互類型的作業(yè)； YARN-Client模式下，Application Master僅僅向YARN請(qǐng)求Executor，Cli

39、ent會(huì)和請(qǐng)求的Container通信來調(diào)度他們工作，也就是說Client不能離開。61杰普軟件科技公司總部：上海市閘北區(qū)萬榮路1188弄龍軟軟件園區(qū)A棟206室：（021）56778147：200436昆山實(shí)訓(xùn)基地：昆山市巴城學(xué)院路828號(hào)昆山浦東軟件園北樓4-5層：（0512）50190290-8000：215311Brighten Your Way And Raise You Up.電郵：training主頁：62Briup Training 1.抽象模型RDD 2.RDD操作構(gòu)建RDDRDD方法閉包基于分區(qū)進(jìn)行操作共享變量普通RDD常見方法鍵值對(duì)RDD特有的方法 3.PageRank案

40、例/ 4.Spark配置與調(diào)優(yōu)評(píng)分?jǐn)?shù)據(jù)分析63杰普軟件科技公司總部：上海市閘北區(qū)萬榮路1188弄龍軟軟件園區(qū)A棟206室：（021）56778147：200436昆山實(shí)訓(xùn)基地：昆山市巴城學(xué)院路828號(hào)昆山浦東軟件園北樓4-5層：（0512）50190290-8000：215311Brighten Your Way And Raise You Up.電郵：training主頁：64Briup Training RDD：彈性分布式數(shù)據(jù)集（Resilient Distributed Dataset） RDD是Spark中最基本的數(shù)據(jù)抽象，它代表一個(gè)不可變(只讀)、可分區(qū)、元 素可并行計(jì)算的,完全彈性

41、的集合。 RDD是有屬性的數(shù)據(jù)集。屬性當(dāng)前數(shù)據(jù)集的狀態(tài)，數(shù)據(jù)集是由數(shù)據(jù)的分區(qū)（partition）組成, 這些分區(qū)運(yùn)行在集群中的不同節(jié)點(diǎn)上 5個(gè)屬性：1. partitions 數(shù)據(jù)分區(qū)列表pute 對(duì)分區(qū)數(shù)據(jù)計(jì)算函數(shù)3.dependencies 依賴列表4.partitioner 分區(qū)方式5.preferredLocations 優(yōu)選位置RDD具有數(shù)據(jù)流模型的特點(diǎn)：自動(dòng)容錯(cuò)、位置感知性調(diào)度和可伸縮性。RDD用戶在執(zhí)行多個(gè)時(shí)顯式地將工作集緩存在內(nèi)存中，后續(xù)的能夠重用工作集，這極大地提升了速度。65Briup Training RDD之所以稱為“ 彈性” 的特點(diǎn) :1. 基于Lineage的高效

42、容錯(cuò)（第n個(gè)節(jié)點(diǎn)出錯(cuò)，會(huì)從第n-1個(gè)節(jié)點(diǎn)恢復(fù)，血 統(tǒng)/緣容錯(cuò)）;2. Task如果失敗會(huì)自動(dòng)進(jìn)行特定次數(shù)的重試（默認(rèn)4次）;3. Stage如果失敗會(huì)自動(dòng)進(jìn)行特定次數(shù)的重試（可以只運(yùn)行計(jì)算失敗的階 段），只計(jì)算失敗的數(shù)據(jù)分片;4. 數(shù)據(jù)調(diào)度彈性：DAG TASK 和5. checkpoint檢查點(diǎn)容錯(cuò);6. 自動(dòng)的進(jìn)行內(nèi)存和磁盤數(shù)據(jù)管理無關(guān);的切換.66Briup Training 對(duì)開發(fā)者而言: 1.RDD可以看作是Spark的一個(gè)對(duì)象，它本身運(yùn)行于內(nèi)存中。 如讀文件是一個(gè)RDD，對(duì)文件計(jì)算是一個(gè)RDD，結(jié)果集也是一個(gè)RDD 不同的分片、 數(shù)據(jù)之間的依賴 、key-value類型的map數(shù)據(jù)

43、都可以看做RDD。 2.RDD不需要物化。 RDD含有如何從其他RDD衍生（即計(jì)算）出本RDD的相關(guān)信息（即Lineage），據(jù)此可以從物理的數(shù)據(jù)計(jì)算出相應(yīng)的RDD分區(qū) 。 RDD作為數(shù)據(jù)結(jié)構(gòu)，本質(zhì)上是一個(gè)只讀的分區(qū)集合。 一個(gè)RDD可以包含多個(gè)分區(qū)，每個(gè)分區(qū)就是一個(gè)dataset片段。 RDD可以相互依賴。 如果RDD的每個(gè)分區(qū)最多只能被一個(gè)Child RDD的一個(gè)分區(qū)使用，則稱之為narrow dependency； 若多個(gè)Child RDD分區(qū)都可以依賴，則稱之為wide dependency。 不同的操作依據(jù)其特性，可能會(huì)產(chǎn)生不同的依賴。例如map操作會(huì)產(chǎn)生narrow depende

44、ncy，而join操作一般則產(chǎn)生wide dependency。67Briup Training68Briup Training69Briup Training70Briup Training Partition結(jié)構(gòu)/* An identifier for a partition in an RDD.*/trait Partition extends Serializable /* Get the partitions index within its parent RDD*/def index: Int/ A better default implementation of HashCode

45、 override def hashCode(): Int = indexoverride def equals(other: Any): Boolean = super.equals(other)71Briup Training72Briup Training 2）compute 對(duì)分區(qū)數(shù)據(jù)計(jì)算函數(shù)。Spark中RDD的計(jì)算是以分區(qū)為的，每個(gè)RDD都會(huì)實(shí)現(xiàn)compute函數(shù)以達(dá)到這個(gè)目的。compute函數(shù)會(huì)對(duì)迭代器 進(jìn)行復(fù)合，不需要保存每次計(jì)算的結(jié)果。 compute(Partition,TaskContext)73Briup Training 3）dependencies 依賴列表： R

46、DD之間的依賴關(guān)系。RDD的每次轉(zhuǎn)換都會(huì)生成一個(gè)新的RDD，所以RDD之間就會(huì)形成類似于流水線一樣的前后依賴關(guān)系。在部分分區(qū)數(shù)據(jù)丟失時(shí)， Spark可以通過這個(gè)依賴關(guān)系重新計(jì)算丟失的分區(qū)數(shù)據(jù)，而不是對(duì)RDD的所有分區(qū)進(jìn)行重新計(jì)算。74Briup Training 1.RDD操作-轉(zhuǎn)化操作與行動(dòng)操作 2.RDD緩存機(jī)制 3.DAG 有向無環(huán)圖(任務(wù)執(zhí)行機(jī)制) 4.lineage/譜系圖/血緣系統(tǒng)(容錯(cuò)機(jī)制) 5.依賴關(guān)系 窄依賴(NarrowDependency):父RDD一個(gè)數(shù)據(jù)分區(qū)只被RDD的一個(gè)數(shù)據(jù)分 區(qū)所使用。filter map等方法 OneToOneDependency RangeD

47、ependency 寬(Shuffle)依賴(ShuffleDependency):父RDD一個(gè)數(shù)據(jù)分區(qū)被子RDD的多個(gè)數(shù)據(jù)分區(qū)所使用。groupByKey sortByKey等方法75Briup Training76Briup Training77Briup Training 4）partitioner 分區(qū)方式： 1.Spark程序通過RDD分區(qū)方式來減少通信開銷 2.只有對(duì)于于key-value的RDD，才會(huì)有Partitioner 3.非key-value的RDD的Parititioner的值是None。 4.Partitioner函數(shù)不但決定了RDD本身的分區(qū)數(shù)量，也決定了paren

48、t RDD Shuffle輸出時(shí)的分區(qū)數(shù)量。 Partitioner源碼實(shí)現(xiàn)abstract class Partitioner extends Serializable def numPartitions: Intdef getPartition(key: Any): Int78Briup Training 1.Spark內(nèi)置支持的兩種分區(qū)方式，一個(gè)是基于哈希的HashPartitioner，另外 一個(gè)是基于范圍的RangePartitioner。 2.查看當(dāng)前RDD分區(qū)方式:rdd.partitioner 3.自定義分區(qū)方式:extends Partitioner重寫兩個(gè)抽象方法即可 4.

49、設(shè)置好RDD分區(qū)的操作: cogroup()、groupWith()、join()、leftOuterJoin()、rightOuterJoin()、groupByKey()、reduceByKey()、combineByKey()、partitionBy(),sortByKey() 以及mapValues()(如果父RDD有分區(qū)方式的話) flatMapValues()(如果父RDD有分 區(qū)方式的話),以及filter()(如果父RDD有分區(qū)方式的話)等，其他操作生成結(jié)果都存在特定分區(qū)方式。 如果RDD在調(diào)用以上操作之前已經(jīng)具有的分區(qū)方式，則以上操作都能夠從分區(qū)中獲益。(此時(shí)以上方法就進(jìn)行數(shù)

50、據(jù)混洗，減少了數(shù)據(jù)混洗的開銷)79Briup Training 5.預(yù)定義自定義分區(qū)(一般預(yù)定義分區(qū)與緩存方法連用) pairRDD.partitionBy(args:Partitioner).cache/persist() 適用場景:迭代計(jì)算中反復(fù)使用同一RDD調(diào)用帶有shuffle操作的方法時(shí)，就 可以在進(jìn)入迭代計(jì)算前，先對(duì)該RDD進(jìn)行預(yù)定義分區(qū)，并記住一定要調(diào)用緩存方法，將該RDD緩存起來，減少數(shù)據(jù)混洗的開銷。 舉例:val newRDD=rdd.partitionBy().cache(); (1 to 50).foreach(x=newRDD.groupByKey.collect /1

51、 次Shuffle) 對(duì)于二元操作，輸出數(shù)據(jù)的分區(qū)方式取決于父RDD的分區(qū)方式。默認(rèn)情況下， 結(jié)果采用哈希分區(qū)，分區(qū)的數(shù)量和并行度一樣。 如果其中一個(gè)父RDD已經(jīng)設(shè)置過分區(qū)方式，那么結(jié)果就會(huì)采用那種分區(qū)方式。 如果兩個(gè)父RDD都設(shè)置過分區(qū)方式，結(jié)果RDD采用第一個(gè)父RDD的分區(qū)方式。80Briup Training 5）preferredLocations 優(yōu)選位置：每個(gè)Partition的優(yōu)先位置的列表（preferred location）。對(duì)于一個(gè)HDFS文件來說，這個(gè)列表保存的就是每個(gè)Partition所在的塊位置。 1.數(shù)據(jù)文件所在塊的位置信息。 2.按照 移動(dòng)數(shù)據(jù)不如移動(dòng)計(jì)算 原則,Spark在進(jìn)行任務(wù)調(diào)度的時(shí)候，會(huì)盡