5.3 Spark编译⽤户可以通过Spark的默认构建⼯具SBT编译和打包源码。当⽤户需要对源码进⾏⼆次开发时需要对源码进⾏增量编译。通过下⾯的⽅式可以实现编译和增量编译。1.克隆Spark源码命令如下git clone https//github.com/apache/spark这样从Github将Spark源码下载到本地建⽴本地的仓库如图5- 14所⽰。2.编译Spark在Spark项⽬的根⽬录内执⾏编译和打包命令命令如下sbt/sbt assembly如图5-15所⽰执⾏过程中会解析依赖和下载需要的依赖Jar包。执⾏完成后将所有Jar包打包为⼀个Jar包即可运⾏Spark集群和⽰例了。3.增量编译在有些情况下⽤户需要修改源码修改之后如果每次都重新下载Jar包或者重新编译⼀遍全部源码就会很浪费时间⽤户可以通过下⾯的增量编译⽅法只编译改变的源码。1编译打包⼀个assembly的Jar包。命令如下。$ sbt/sbt clean assembly2这时的Spark程序已经可以运⾏。⽤户可以进⼊Spark Shell执⾏程序。命令如下$ ./bin/spark-shell3配置export SPARK_PREPEND_CLASSES参数为true开启增量编译模式。命令如下$ export SPARK_PREPEND_CLASSEStrue4继续使⽤Spark Shell中的程序。通过下⾯命令进⼊SparkShell。$ ./bin/spark-shell5这时可以对代码进⾏修改和⼆次开发。/*开发……*/6编译Spark源码。$ sbt/sbt compile/*开发……*/$ sbt/sbt compile7解除增量编译模式。$ unset SPARK_PREPEND_CLASSES8返回正常使⽤Spark Shell的情景。通过下⾯命令进⼊SparkShell$ ./bin/spark-shell # Back to normal using Spark classes from the assembly jar9如果不想每次都开启⼀个新的sbt会话就可以在compile命令前加上~。命令如下$ sbt/sbt ~compile4.查看Spark源码依赖图使⽤SBT查看依赖图需要运⾏下⾯的命令结果如图5-16所⽰。$ # sbt$ sbt/sbt dependency-tree使⽤Maven查看依赖图需要运⾏下⾯的命令。$ # Maven$ mvn -DskipTests install $ mvn dependencytree5.4 配置Spark源码阅读环境由于Spark使⽤SBT作为项⽬管理构建⼯具SBT的配置⽂件中配置了依赖的jar包⽹络路径在编译或者⽣成指定类型项⽬时需要从⽹络下载jar包需要预先安装Git。在Linux操作系统或者Windows操作系统中可以下载Git Shell在Git Shell中进⾏命令⾏操作通过sbt/sbt gen-idea命令⽣成Intellij项⽬⽂件然后在Intellij IDEA中直接通过“Open Project”选项打开项⽬。1克隆Spark源码。命令如下git clone https//github.com/apache/spark2⽣成项⽬。3在所需的软件安装好后在Spark源代码根⽬录下输⼊以下代码。sbt/sbt gen-idea这样SBT会⾃动下载依赖包和编译源⽂件编译并⽣成Intellij所需的项⽬⽂件。如果在Eclipse下则需要运⾏如下代码sbt/sbt gen-eclipse这样SBT会⾃动下载依赖包和编译源⽂件编译并⽣成Eclipse所需的项⽬⽂件。⽤户就可以在IDE中查看源码了。5.5 本章⼩结本章主要介绍了Spark应⽤程序的开发流程以及如何编译和调试Spark程序。⽤户可以选⽤能够很好⽀持Scala项⽬的Intellij IDE。如果之前经常使⽤Eclipse开发Java程序也可以在Eclipse中安装Scala IDE插件开发与调试Spark程序。由于Spark项⽬基于SBT构建⽤户可以创建SBT项⽬开发应⽤。在应⽤的开发过程中需要进⾏调试诊断问题。在本章最后部分介绍的远程调试⽅法可以很好地帮助⽤户调试Spark程序。通过本章的介绍读者可以搭建Spark开发环境下⾯将通过Spark编程实战进⼊Spark程序的开发之旅。第6章 Spark编程实战前⾯章节已经介绍了很多关于Spark的基础知识、运⾏机制 Spark的集群安装与配置⽅法Spark内核是由Scala语⾔开发的当然⽤户也可以使⽤Java和Python进⾏开发。本章将从题⽬、程序和应⽤出发介绍如何使⽤Spark进⾏程序开发以及如何编写Spark程序。6.1 WordCountWordCount是⼤数据领域的经典范例如同程序设计中的HelloWorld⼀样是⼀个⼊门程序。本节主要从并⾏处理的⾓度出发介绍设计Spark程序的过程。1.实例描述输⼊Hello World Bye WorldHello Hadoop Bye HadoopBye Hadoop Hello Hadoop输出Bye3Hadoop4Hello3World22.设计思路在map阶段会将数据映射为Hello1World1Bye1World1Hello1Hadoop1Bye1Hadoop1Bye1Hadoop1Hello1Hadoop1在reduceByKey阶段会将相同key的数据合并并将合并结果相加。Bye111Hadoop1111Hello111World113.代码⽰例WordCount的主要功能是统计输⼊中所有单词出现的总次数编写步骤如下。1初始化创建⼀个SparkContext对象该对象有4个参数Spark master位置、应⽤程序名称、Spark安装⽬录和Jar存放位置。需要引⼊下⾯两个⽂件。import org.apache.spark._import SparkContext._val sc new SparkContextargs0 WordCountSystem.getenvSPARK_HOMESeqSystem.getenvSPARK_TEST_JAR2加载输⼊数据从HDFS上读取⽂本数据可以使⽤SparkContext中的textFile函数将输⼊⽂件转换为⼀个RDD该函数采⽤Hadoop中的TextInputFormat解析输⼊数据。val textRDD sc.textFileargs1textFile中的每个Hadoop Block相当于⼀个RDD分区。3词频统计对于WordCount⽽⾔⾸先需要从输⼊数据中的每⾏字符串中解析出单词然后分⽽治之将相同单词放到⼀个组中统计每个组中每个单词出现的频率。val result textRDD.flatMap{casekey value value.toString.split\\s}.mapword word 1. reduceByKey _ _其中flatMap函数每条记录转换转换后如果每个记录是⼀个集合则将集合中的元素变为RDD中的记录map函数将⼀条记录映射为另⼀条记录reduceByKey函数将key相同的关键字的数据聚合到⼀起进⾏函数运算。4存储结果可以使⽤SparkContext中的saveAsTextFile函数将数据集保存到HDFS⽬录下。result.saveAsSequenceFileargs24.应⽤场景WordCount的模型可以在很多场景中使⽤如统计过去⼀年中访客的浏览量、最近⼀段时间相同查询的数量和海量⽂本中的词频等。6.2 Top KTop K算法有两步⼀是统计词频⼆是找出词频最⾼的前K个词。1.实例描述假设取Top 1则有如下输⼊和输出。输⼊Hello World Bye WorldHello Hadoop Bye HadoopBye Hadoop Hello Hadoop输出词Hadoop 词频42.设计思路⾸先统计WordCount的词频将数据转化为词词频的数据对第⼆个阶段采⽤分治的思想求出RDD每个分区的Top K最后将每个分区的Top K结果合并以产⽣新的集合在集合中统计出Top K的结果。每个分区由于存储在单机的所以可以采⽤单机求Top K的⽅式。本例采⽤堆的⽅式。也可以直接维护⼀个含K个元素的数组感 兴趣的读者可以参考其他资料了解堆的实现。 3.代码⽰例 Top K算法⽰例代码如下 import org.apache.spark.SparkContext import org.apache.spark.SparkContext._ object TopK { def mainargsArray[String] { /*执⾏WordCount统计出最⾼频的词*/ val spark new SparkContextlocal TopK System.getenvSPARK_HOME SparkContext.jarOfClassthis.getClass val count spark.textFiledata.flatMapline line.split .mapword word 1.reduceByKey_ _ /*统计RDD每个分区内的Top K查询*/ val topk count.mapPartitionsiter { whileiter.hasNext { putToHeapiter.next } getHeap.iterator } .collect /*将每个分区内统计出的TopK查询合并为⼀个新的集合统计出TopK查询*/ val iter topk.iterator whileiter.hasNext { putToHeapiter.next } val outitergetHeap.iterator /*输出TopK的值*/ printlnTopk 值 whileoutiter.hasNext { println\n 词频outiter.next._1 词outiter.next._2 } spark.stop } } def putToHeapiter String Int { /*数据加⼊含k个元素的堆中*/ …… } def getHeap Array[String Int] { /*获取含k个元素的堆中的元素*/ val anew Array[String Int] …… }4.应⽤场景 Top K的⽰例模型可以应⽤在求过去⼀段时间消费次数最多的消费 者、访问最频繁的IP地址和最近、更新、最频繁的微博等应⽤场景。6.3 中位数海量数据中通常有统计集合中位数的计算需求读者可以通过以下⽰例了解Spark求中位数的⽅式。1.实例描述若有很⼤⼀组数据数据的个数是N在分布式数据存储情况下找到这N个数的中位数。数据输⼊是以下整型数据。1、2、3、4、5、6、8、9、11、12、34输出为62.设计思路海量数据求中位数有很多解决⽅案。假设海量数据已经预先排序本例的解决⽅案为将整个数据空间划分为K个桶。第⼀轮在mapPartition阶段先将每个分区内的数据划分为K个桶统计桶中的数据量然后通过reduceByKey聚集整个RDD每个桶中的数据量。第⼆轮根据桶统计的结果和总的数据量可以判读数据落在哪个桶⾥以及中位数的偏移量offset。针对这个桶的数据进⾏排序或者采⽤Top K的⽅式获取到偏移为offset的数据。3.代码⽰例import org.apache.spark.{SparkContext SparkConf} import org.apache.spark.SparkContext.rddToPairRDDFunctions */ object Median { def mainargs Array[String] { val conf new SparkConf.setAppNameSpark Pi val spark new SparkContextconf val data spark.textFiledata /*将数据逻辑划分为10个桶这⾥⽤户可以⾃⾏设置桶数量统计每个桶中落⼊的数据量*/ val mappeddata data.mapnum { num/1000 num } val count mappeddata.reduceByKeya b { ab }.collect /*根据总的数据量逐次根据桶序号由低到⾼依次累加判断中位数落在哪个 桶中并获取到中位数在桶中的偏移量*/ val sum_countcount.mapdata { data._2 }.sum var temp 0 var index 0 var mid sum_count/2 for i - 0 to 10 { temptempcounti iftemp mid { indexi break } } /*中位数在桶中的偏移量*/ val offset temp - mid /*获取到中位数所在桶中的偏移量为offset的数也就是中位数*/ val result mappeddata.filternum num._1 index .takeOrdered offset printlnMedian is resultoffset spark.stop } }4.应⽤场景统计海量数据时经常需要预估中位数由中位数⼤致了解某列数据做机器学习和数据挖掘的很多公式中也需要⽤到中位数。6.4 倒排索引倒排索引inverted index源于实际应⽤中需要根据属性的值来查找记录。在索引表中每⼀项均包含⼀个属性值和⼀个具有该属性值的各记录的地址。由于记录的位置由属性值确定⽽不是由记录确定因⽽称为倒排索引。将带有倒排索引的⽂件称为倒排索引⽂件简称倒排⽂件inverted file。其基本结构如图6-1所⽰。搜索引擎的关键步骤是建⽴倒排索引。相当于为互联⽹上⼏千亿⻚⽹⻚做了⼀个索引与书籍⽬录相似⽤户想看与哪⼀个主题相关的章节直接根据⽬录即可找到相关的⻚⾯。1.实例描述输⼊为⼀批⽂档集合合并为⼀个HDFS⽂件以分隔符分隔。Id1 The Spark……Id3 The Spark……输出如下单词⽂档ID合并字符串。Spark id1 id2Hadoop id3 id4The id1 id3 id62.设计思路⾸先进⾏预处理和分词转换数据项为⽂档ID⽂档词集合的RDD然后将数据映射为词⽂档ID的RDD去重最后在reduceByKey阶段聚合每个词的⽂档ID。3.代码⽰例倒排索引的⽰例代码如下所⽰import org.apache.spark.SparkContext import scala.collection.mutable._ object InvertedIndex { def mainargs Array[String] { val spark new SparkContextlocal TopK System.getenvSPARK_HOME SparkContext.jarOfClassthis.getClass /*读取数据数据格式为⼀个⼤的HDFS⽂件中⽤\n分隔不同的⽂件⽤\t分隔⽂件ID和⽂件内容⽤ 分隔⽂件内的词汇*/ val words spark.textFiledir.mapfile file.split\t.mapitem { item0 item1 }.flatMapfile { /*将数据项转换为LinkedList[词 ⽂档id]的数据项并通过flatmap将RDD内的数据项转换 为词 ⽂档ID*/ val list new LinkedList[String String] val words file._2.split .iterator whilewords.hasNext { listwords.next } list }.distinct /*将词⽂档ID的数据进⾏聚集相同词对应的⽂档ID统计到⼀起形成*/ 词 ⽂档ID1⽂档ID2 ⽂档ID3……形成简单的倒排索引*/ words.mapword { word._2 word._1 }.reducea b { a\tb }.saveAsTextFileindex } }4.应⽤场景搜索引擎及垂直搜索引擎中需要构建倒排索引⽂本分析中有的场景也需要构建倒排索引。