《计算及其实践教程（第二版）》第六章Spark平台教学课件

大数据平台Spark插上翅膀的大象（SparkonHadoop）内容三种计算框架Spark产生背景Spark特点Spark生态系统Spark核心概念RDDSpark程序设计实例进一步理解Spark核心概念RDD三种计算框架根据伯克利大学提出的关于数据分析的软件栈BDAS（BerkeleyDataAnalyticsStack），目前的大数据处理可以分为如以下三个类型

批处理计算流式计算交互式计算批处理(Batch)计算成批处理数据，特点是吞吐量大，但处理速度慢，实时性差。在三种计算框架中实时性最低，响应时间在分钟级到数十分钟级，有时甚至达到数小时。批处理计算和其它两种计算模式的差别就好比火车和飞机的差别，火车一次载人多，吞吐量大，但速度较慢；飞机一次载人少，但实时性好。典型实例：MapReduce、Hive、Pig适用场合：适合PB级以上海量数据的离线处理。比如MapReduce的输入数据必须是静态的（即离线的），不能动态变化。很多搜索引擎在统计过去一年或半年时间跨度内的最流行的K个搜索词时用到基于批处理的MapReduce计算框架（历史数据分析而不是实时数据分析）。流式计算(Streaming)快速实时小批量地处理数据。在三种计算框架中实时性最高，响应时间在数百毫秒级到数秒级。典型实例：Storm、SparkStreaming。适用场合：适合处理大量在线流式数据，并返回实时的结果。比如在电子商务中统计过去1分钟内访问最多的5件商品；淘宝双11（光棍节）时实时统计网站商品的总交易额；社交网络趋势追踪；网站指标统计、点击日志分析等。交互式计算(Interactive)以近实时方式处理SQL或类SQL交互式数据。在三种计算框架中实时性居中，响应时间在数十秒到数分钟之间。典型实例：Impala、SparkSQL。适用场合：适合以请求-响应的交互方式处理大量结构化和半结构化数据。比如使用SQL查询结构化的数据，很容易地完成包括商业智能BI在内的各种复杂的数据分析算法。内容三种计算框架Spark产生背景Spark特点Spark生态系统Spark核心概念RDDSpark程序设计实例进一步理解Spark核心概念RDDSpark产生背景：MapReduce局限性MapReduce框架局限性仅支持Map和Reduce两种操作迭代计算效率低（如不适应机器学习、图计算等需要大量迭代的计算）不适合交互式处理（如数据挖掘）不适合流式处理（如点击日志分析）MapReduce编程不够灵活尝试scala函数式编程语言Spark产生背景：框架多样化现有的各种计算框架各自为战批处理：MapReduce、Hive、Pig流式计算：Storm交互式计算：Impala能否有一种灵活的框架可同时进行批处理、流式计算、交互式计算等？产生背景：大统一系统在一个统一的框架下，进行批处理、流式计算、交互式计算内容三种计算框架Spark产生背景Spark特点Spark生态系统Spark核心概念RDDSpark程序设计实例进一步理解Spark核心概念RDDSpark特点高效内存计算引擎DAG图比MapReduce快10~100倍易用提供了丰富的API，支持Java，Scala，Python等多种语言同一功能实现，Scala代码量比MapReduce少2~5倍与Hadoop集成读写HDFS/Hbase与YARN集成Spark特点1：高效—MapReduce慢是由于MapReduce用磁盘存放中间结果，而且写3份，用户的查询请求需要多次频繁地访问某个数据集时磁盘I/O慢，数据的复制和序列化也导致运行速度慢。Spark特点1：高效—SparkSpark快的原因是把输入数据一次性读取到分布式内存中，以后用户的的多次查询和处理都从内存中读写，其速度是MapReduce网络复制和磁盘I/O速度的10到100倍。由于技术的进步，机器的内存空间越来越大，而且内存条也越来越便宜，使Spark这种基于内存的计算成为可能。但当数据量很大时，内存空间若放不下，Spark有相关机制可以把热点数据（比如一天读取几百次的数据）放入内存而把非热点数据（比如一个月读取一次的数据）放入磁盘。Spark特点1：高效—Spark当迭代次数达到30次时，Hadoop的运行时间大约是4000s，而Spark的运行时间大约是400s，相差10倍左右。Spark特点2：易用—MapReduceSpark特点2：易用—SparkSpark特点3：与Hadoop集成内容三种计算框架Spark产生背景Spark特点Spark生态系统Spark核心概念RDDSpark程序设计实例进一步理解Spark核心概念RDDSpark生态系统Alluxio(原名Tachyon)定义：Tachyon(超光子)是一种内存式的文件系统，可以认为是搭建在HDFS上的分布式缓存。它可以在集群里以访问内存的速度来访问存在Tachyon里的文件。架构：Tachyon是架构在最底层的分布式文件存储和上层的各种计算框架之间的一种中间件。具体地说，是在分布式文件存储系统如HDFS、S3等之上，在Spark、MapReduce、Impala等各种计算框架之下。引入Tachyon的原因1）提高不同任务或框架间数据交换的速度不同任务或不同计算框架间的数据共享情况在所难免，例如Spark的分属不同Stage的两个任务，或Spark与MapReduce框架的数据交互。在这种情况下，一般就需要通过磁盘来完成数据交换，而这通常是效率很低的。而引入Tachyon中间层后，数据交换实际上是在内存中进行的。引入Tachyon的原因2）使Spark的执行引擎和存储引擎分开Spark作为内存计算框架，为什么还需要再加一层内存管理的文件系统？因为Spark其实只提供了强大的内存计算能力，但未提供存储能力。那么默认让Spark自己直接在内存管理数据不行吗？让Spark自己来管理内存会出现的问题。默认情况下，Spark的任务执行和数据本身都在一个进程内。当执行出现问题时就会导致整个进程崩溃，并丢失进程内的所有数据。而Tachyon这一中间层的引入，就相当于将存储引擎从Spark中抽离出来，从而每个任务进程只负责执行。进程的崩溃不会丢失数据，因为数据都在Tachyon里面了。引入Tachyon的原因3）避免数据被重复加载不同的Spark任务可能会访问同样的数据，例如两个任务都要访问HDFS中的某些Block。这时每个任务都要自己去磁盘加载数据到内存中。而Tachyon可以只保存一份数据在内存中供加载，而且它还使用堆外内存，避免GC（垃圾收集）开销。Mesos和YARNMesos是一个开源的资源管理系统，可以对集群中的资源做弹性管理。目前Twitter,Apple等公司在使用Mesos管理集群资源。Apple的siri的后端便是采用Mesos进行资源管理。目前看来，HadoopYARN要比Mesos更主流，前景更广阔。YARN在实现资源管理的前提下，能够跟Hadoop生态系统完美结合。YARN定位为大数据中的数据操作系统，能够更好地为上层各类应用程序（MapReduce/Spark）提供资源管理和调度功能。另外，非常重要的一点是，YARN的社区力量要比Mesos强大的多，它的参与人员众多，周边系统的建设非常完善。Shark和SparkSQLSparkSQL是分布式SQL查询引擎，用于处理交互式数据流，把SQL命令分解成多个任务（Task）交给Hadoop集群处理。自2013年3月面世以来，SparkSQL已经成为除SparkCore以外最大的Spark组件。在2014年7月1日的SparkSummit上，Databricks公司宣布终止对Shark的开发，将重点放到SparkSQL上。SparkSQL将涵盖Shark的所有特性，用户可以从Shark0.9进行无缝的升级。除了接过Shark的接力棒，继续为Spark用户提供高性能的SQLonHadoop解决方案之外，SparkSQL还为Spark带来了通用、高效、多元一体的结构化数据处理能力。SparkSQL可加载和查询各种数据源，比如Hive数据、Parquet列式存储格式数据、JSON格式数据、通过JDBC和ODBC等连接各种数据源SparkStreamingSparkStreaming是大规模流式数据处理的新贵，将流式计算分解成一系列短小的批处理作业。SparkStreaming类似于ApacheStorm，用于流式数据的处理。SparkStreaming有高吞吐量和容错能力强这两个特点。SparkStreaming支持的数据输入源很多，例如：HDFS、Kafka、Flume、Twitter、ZeroMQ和简单的TCP套接字等。数据输入后可以用Spark的高度抽象原语如map、reduce、join、window等进行运算。而结果也能保存在很多地方，如HDFS、数据库等。另外SparkStreaming也能和MLlib（机器学习）以及GraphX完美融合。SparkStreaming支持的数据输入源和输出地GraphXSparkGraphX是一个分布式图处理框架，SparkGraphX基于Spark平台提供对图计算和图挖掘简洁易用的而丰富多彩的接口，极大的方便了大家对分布式图处理的需求。社交网络中人与人之间有很多关系链，例如Twitter、Facebook、微博、微信，这些都是大数据产生的地方，都需要图计算，现在的图处理基本都是分布式的图处理，而并非单机处理。图的分布式或者并行处理其实是把这张图拆分成很多的子图，然后我们分别对这些子图进行计算，计算的时候可以分别迭代进行分阶段的计算，即对图进行并行计算。GraphX使用的是一种点和边都带有属性的有向多重图，有Table和Graph两种视图，只需一份物理存储。MLBaseMLBase是Spark生态圈的一部分，专注于机器学习，包含三个组件：MLlib、MLI、MLOptimizer。MLlib是Spark的机器学习库，包括分类、聚类、回归算法、决策树、推荐算法等各种机器学习的核心算法的实现。MLI是MLlib的测试床，是用于特征提取和算法开发的实验性API。MLBase的核心是MLOptimizer，把声明式的任务转化成复杂的学习计划，输出最优的模型和计算结果。MLBase与其它机器学习系统Mahout的不同：2014年4月，Mahout告别MapReduce实现，转而采用Spark作为底层实现基础。MLBase是自动化的，Mahout需要使用者具备机器学习技能，来选择自己想要的算法和参数来做处理。MLBase提供了不同抽象程度的接口，可以扩充ML（机器学习）算法。内容三种计算框架Spark产生背景Spark特点Spark生态系统Spark核心概念RDDSpark程序设计实例进一步理解Spark核心概念RDDSpark核心概念—RDDRDD(ResilientDistributedDatasets，弹性分布式数据集)分布在集群中的只读对象集合，由多个Partition(分区)构成，可以存储在磁盘或内存中（有多种存储级别）。可以由Scala集合创建RDD，比如sc.parallelize(List(1,2,3))//对集合List(1,2,3)进行并行化生成RDD，默认Partition数量为1sc.parallelize(1ton,5)))//对集合1到n（n可以是很大的整数）进行并行化生成RDD，指定Partition数量为5利用本地文件或HDFS文件创建RDD1.由文本文件（TextInputFormat）创建RDDsc.textFile(“file.txt”)//将本地文本文件加载成RDDsc.textFile(“directory/*.txt”)//将某类文本文件加载成RDDsc.textFile(“hdfs://nn:9000/path/file”)//将hdfs文件或目录加载成RDDsc.textFile(“file:///file.txt”)//将本地文本文件加载成RDD，用file:///指定本地文件2.由sequenceFile文件（SequenceFileInputFormat）创建RDDsc.sequenceFile(“file.txt”)//将本地二进制文件加载成RDDsc.sequenceFile[String,Int](“hdfs://nn:9000/path/file”)//将hdfs二进制文件加载成RDDRDDRDD是只读数据集，不支持修改，不考虑并发和加锁。那么如何对RDD进行操作呢？有两种基本操作（operator）：Transformation（转换）Action（行动）TransformationTransformation（转换）可通过Scala集合或者Hadoop数据文件构造一个新的RDD通过已有的RDD产生新的RDD（输入是RDD，输出还是RDD，则为Transformation）举例：map，filter，groupBy，reduceByAction（行动）通过RDD计算得到一个或者一组值（输入是RDD，输出不是RDD，而是一个或一组值或写入存储，则为Action）举例：count，collect，saveTransformation操作和Action操作的重要区别：无论执行了多少次Transformation操作，RDD都不会真正执行运算（只做标记，不触发计算），只有当Action操作被执行时，运算才会触发。这也称为Spark的惰性执行，即一段Spark代码不会执行，直到遇到第一个Action惰性执行的好处是优化的概率提高，即看到的步骤越多，最终执行时优化的可能性就越高。两段代码执行效果有何不同beacons=spark.textFile(“hdfs://...”)cachedBeacons=beacons.cache()cachedBeacons.filter(_.contains(“HouseOfCards”))cachedBeacons.filter(_.contains(“GameOfThrone”))……beacons=spark.textFile(“hdfs://...”)cachedBeacons=beacons.cache()cachedBeacons.filter(_.contains(“HouseOfCards”)).countcachedBeacons.filter(_.contains(“GameOfThrone”)).count……这两段代码都从HDFS文件中生成RDD，然后用cache操作把RDD数据缓存到内存，接着执行了两个filter操作，对缓存到内存中的RDD数据进行过滤，找出包含特定字符串的数据。但第二段代码在filter操作之后用了count操作，注意到count是Action操作，所以会真正触发计算，从而得到包含特定字符串的数据个数。Operator示例RDD1包含集合1到7的整数，由3个Partition组成。经过第一个map(+1)操作，生成RDD2，其中集合的每个整数加1，但该操作的结果并不真的生成，而是只做了标记，不在内存或磁盘生成该中间结果。因为map是Transformation操作。RDD2接着执行SAVEASTEXTFILE(“HDFS://…”)操作，该操作是Action操作，因此计算被触发，把RDD中的加1后的数据保存到存储系统HDFS指定的文件中。RDDtransformation举例1）//创建RDD，用集合创建RDDvalnums=sc.parallelize(List(1,2,3))2）//将RDD传入map函数，求变量的平方，生成新的RDDvalsquares=nums.map(x=>x*x)//{1,4,9}3）//对RDD中元素进行过滤，求模2结果为0的偶数，生成新的RDDvaleven=squares.filter(_%2==0)//{4}4）//利用flatMap将一个元素映射成多个，生成新的RDDnums.flatMap(x=>1tox)//=>{1,1,2,1,2,3}RDDAction举例1）//用集合创建新的RDDvalnums=sc.parallelize(List(1,2,3))2）//将RDD保存为本地数组，collect将分布式的RDD返回为一个单机的数组，返回到driver程序所在的机器nums.collect()//=>Array(1,2,3)3）//返回前K个元素，返回一个数组，该操作目前并非并行执行，而是由driver程序所在机器执行的nums.take(2)//=>Array(1,2)4）//计算元素总数nums.count()//=>35）//合并集合元素nums.reduce(_+_)//=>66）//将RDD写到HDFS中nums.saveAsTextFile(“hdfs://nn:8020/output”)nums.saveAsSequenceFile(“hdfs://nn:8020/output”)Key/Value类型的RDD1）valpets=sc.parallelize(List((“cat”,1),(“dog”,1),(“cat”,2)))2）pets.reduceByKey(_+_)//=>{(cat,3),(dog,1)}3）pets.groupByKey()//=>{(cat,Seq(1,2)),(dog,Seq(1)}4）pets.sortByKey()//=>{(cat,1),(cat,2),(dog,1)}Transformation和Action内容三种计算框架Spark产生背景Spark特点Spark生态系统Spark核心概念RDDSpark程序设计实例进一步理解Spark核心概念RDD实例1：wordcount//导入相关类，_相当于*号，即导入一个包中所有的类importorg.apache.spark._importSparkContext._//定义对象WordCountobjectWordCount{//定义main函数，冒号后为数据类型defmain(args:Array[String]){//如果输入的参数个数不等于3，则打印输入命令格式的提示信息，//参数0为master地址，参数1为输入数据所在目录，比如://hdfs://host:port/input/data，参数2为数据输出目录，比//如:hdfs://host:port/output/dataif(args.length!=3){println("usageisorg.test.WordCount<master><input><output>")return}实例1：wordcount//创建SparkContext对象，封装了spark执行环境信息，依次包括//master地址、作业名称、Spark安装目录、作业依赖的jar包val

sc=newSparkContext(args(0),"WordCount",System.getenv("SPARK_HOME"),Seq(System.getenv("SPARK_TEST_JAR")))//用输入的文本文件创建RDDval

textFile=sc.textFile(args(1))//单词计数的程序主体。其中split根据给定的正则表达式的匹配拆分//字符串,\s表示空格、回车、换行等分隔符，+表示一个或多个的意//思valresult=textFile.flatMap(line=>line.split("\\s+")).map(word=>(word,1)).reduceByKey(_+_)//把单词计数的结果存放到输出文件中result.saveAsTextFile(args(2))}}Spark程序设计—基本流程1)创建SparkContext对象这是Spark程序的入口，封装了spark执行环境信息。有点类似数据库编程中的建立连接操作。2)创建RDD可从Scala集合或Hadoop数据文件上创建3)在RDD之上进行转换和actionSpark提供了多种转换和action函数4)返回结果保存到HDFS中，或直接打印出来Spark程序设计—Scala用函数式编程的方式处理集合：1)varlist=List(1,2,3)//“=>”操作的含义可理解为把左边的变量装到右边的表达式中2)list.foreach(x=>println(x))//打印1，2，3list.foreach(println)//与上式等价，形式更简练3)list.map(x=>x+2)//=>List(3,4,5)list.map(_+2)//与上式等价。此处“_”称为占位符，对集合中的每个元素作用一次。4)list.filter(x=>x%2==1)//=>List(1,3)list.filter(_%2==1)//与上式等价。5)list.reduce((x,y)=>x+y)//6list.reduce(_

+_)//与上式等价。Wordcount程序运行过程实例实例2：分布式估算Pi-蒙特卡洛算法它的具体定义是：在广场上画一个边长一米的正方形，在正方形内部随意用粉笔画一个不规则的形状，现在要计算这个不规则图形的面积，怎么计算呢?蒙特卡洛(MonteCarlo)方法告诉我们，均匀的向该正方形内撒N（N是一个很大的自然数）个黄豆，随后数数有多少个黄豆在这个不规则几何形状内部，比如说有M个，那么，这个奇怪形状的面积便近似于M/N，N越大，算出来的值便越精确。在这里我们要假定豆子都在一个平面上，相互之间没有重叠。蒙特卡洛方法可用于近似计算圆周率公式推导1）假设正方形边长为d,则：正方形面积为：d*d圆的面积为：pi*(d/2)*(d/2)正方形与圆两者面积之比为：4/pi2）随机产生位于正方形内的点n个，假设落到圆中的有count个，则：Pi=4*count/n3）当n->∞时，Pi逼近真实值分布式估算Pi—调用“SparkPi”程序//定义对象SparkPiobjectSparkPi{//定义函数maindefmain(args:Array[String]){//创建SparkConf，封装了Spark配置信息：应用程序名称valconf=newSparkConf().setAppName("SparkPi")//创建SparkContext,封装了调度器等信息valspark=newSparkContext(conf)分布式估算Pi--计算//启动一定数量的maptask进行并行处理，默认数量为2valslices=if(args.length>0)args(0).toIntelse2//初始化nvaln=100000*slices//计算落入园内的点数valcount=spark.parallelize(1ton,slices).map{i=>valx=random*2-1valy=random*2-1if(x*x+y*y<1)1else0}.reduce(_+_)分布式估算Pi—输出//根据公式输出Pi值println("Piisroughly"+4.0*count/n)//程序结束spark.stop()}}分布式估算Pivalcount=spark.parallelize(1ton,slices).map{i=>valx=random*2-1valy=random*2-1if(x*x+y*y<1)1else0}.reduce(_+_)对集合1到n中的每个元素执行map内含的匿名函数，该匿名函数基于以原点为圆心的坐标系，半径为1，直径为2，random函数给出一个(0,1)范围内的随机数，x和y是坐标（取值范围为-1到1），根据圆的方程，点落在圆内则map的结果为1，否则为0。Reduce函数对map的结果即一系列1和0进行累加，求出落入园内的点数。程序架构程序架构几个基本概念：（1）job：包含多个task组成的并行计算，往往由action催生。（2）stage：job的调度单位。（3）task：被送到某个executor上的工作单元。（4）taskSet：一组关联的，相互之间没有shuffle依赖关系的任务组成的任务集。一个应用程序由一个driverprogram和多个job构成。一个job由多个stage组成。一个stage由多个没有shuffle关系的task组成。task是计算的最小单位。几个基本概念程序架构Driver:运行Application的main()函数并创建SparkContext。通常用SparkContext代表Driver。Spark程序架构采用主从结构，一个Driver负责程序入口，多个executor分布式执行更多个task。一个executor上可运行多个task，一个task相当于一个线程，多个executor共享一个WorkerNode.程序架构spark应用程序的运行架构：由driver向集群申请资源，集群分配资源，启动executor。driver将spark应用程序的代码和文件传送给executor。executor上运行task，运行完之后将结果返回给driver或者写入HDFS。在实例1的WordCount中，task的数目即任务总量由输入HDFS的文件划分的block数决定，一个block的大小默认为64MB（Hadoop2.0默认为128MB），HDFS的一个block默认为SparkRDD中的一个Partition，即一个task处理的数据量。同样地，在实例2中，SparkRDD中的一个Partition即一个task处理的数据量，所以语句spark.parallelize(1ton,slices)中slices变量的值为指定的task数，也是指定的Partition数。体验spark：交互式模式spark-shellSpark-shell交互式运行spark代码，类似于scala交互式终端可用于快速体验spark，查看程序运行结果等如何使用bin/spark-shell.shscala>valdata=Array(1,2,3,4,5)//产生datascala>valdistData=sc.parallelize(data)//将data处理成RDDscala>distData.reduce(_+_)提交Spark程序（运行在YARN上）exportYARN_CONF_DIR=/opt/hadoop/yarn-client/etc/hadoopbin/spark-submit\--masteryarn-cluster\--classcom.hulu.examples.SparkPi\--namesparkpi\--driver-memory2g\--executor-memory3g\--executor-cores2\--num-executors2\--queuespark\--confspark.pi.iterators=500000\--confspark.pi.slices=10\$FWDIR/target/scala-2.10/spark-example-assembly-1.0.jarMaster参数指定程序运行模式，比如local,yarn-client,yarn-cluster。该参数必选。class参数指定应用程序主类。该参数必选。name参数指定作业名称。该参数可选。driver-memory参数指定Driver需要的内存。该参数可选，默认为512MBexecutor-memory参数指定每个executor需要的内存。该参数可选，默认为1GBexecutor-cores参数指定每个executor线程数，相当于每个executor中的task数。该参数可选，默认为1num-executors参数指定需启动的Executor总数。该参数可选，默认为2queue参数指定提交应用程序给哪个YARN队列。该参数可选，默认为default队列。conf参数指定用户自定义配置。最后一行指定用户应用程序所在jar包，必选，且一定要放在整个spark-submit提交命令的最后。提交Spark程序（运行在YARN上）内容三种计算框架Spark产生背景Spark特点Spa