Spark各个知识点总结讲解PPT课件.ppt

单击此处编辑母版标题样式,单击此处编辑母版文本样式,第二级,第三级,第四级,第五级,*,单击此处编辑母版文本样式,第二级,*,单击此处编辑母版标题样式,单击此处编辑母版文本样式,第二级,*,单击此处编辑母版标题样式,Spark简介,1,Spark,是什么,Spark是一个,快速,且,通用,的,集群计算,平台。,集群计算,把一台电脑无法解决的问题，放到多台电脑组成的集群上进行解决，这就是集群计算。,2,Spark,的特点,Spark是快速的,很多任务能够秒级完成，对于,一些特定的工作,，Spark比Mapreduce快10-20倍。,Spark扩充了流行的Mapreduce计算模型，使Spark更高效地支持更多类型的计算，包括交互式查询，和流处理。,速度快的另一个主要原因就是，,能够在内存中计算,。,3,Spark,的特点,Spark是通用的,Spark的设计，容纳了之前很多独立的，分布式系统所拥有的功能。独立的分布式系统包括：批处理，迭代式计算，交互查询和流处理等。,并且，由之前需要维护不同的集群，到现在只需要维护一个Spark集群。,4,Spark,的特点,Spark是高度开放的,Spark提供了Python，Java，Scala，SQL的API和丰富的内置库。,同时，Spark和其它的大数据工具整合的很好。尤其，Spark能够运行在Hadoop集群上面，能够访问Hadoop数据。,5,Spark,的组件,Spark包括多个,紧密集成,的组件。,6,Spark,的组件,紧密集成的优点：,如果Spark底层优化了，那么基于Spark底层的组件，也得到了相应的优化。例如，Spark底层增加了一个优化算法，那么Spark的SQL和机器学习包也会自动的优化。,紧密集成，节省了各个组件组合使用时的部署，测试等时间。,当向Spark增加新的组件时，其它的组件，可以立刻享用新组件的功能。,无缝连接不同的处理模型。,7,Spark,的组件,Spark Core：,包含Spark的基本功能，包含任务调度，内存管理，容错机制等。,Spark Core内部定义了RDDs(resilient distributed datasets，弹性分布式数据集)。,RDDs代表横跨很多工作节点的数据集合，RDDs可以被并行的处理。,Spark Core提供了很多APIs来创建和操作这些集合(RDDs)。,8,Spark,的组件,Spark SQL：,是Spark处理结构化数据的库。它支持通过SQL查询数据，就像HQL(Hive SQL)一样，并且支持很多数据源，像Hive表，JSON等。Spark SQL是在Spark 1.0版本中新加的。,Shark是一种较老的基于Spark的SQL项目，它是基于Hive修改的，它现在已经被Spark SQL替代了。,9,Spark,的组件,Spark Streaming:,是实时数据流处理组件，类似Storm。,Spark Streaming提供了API来操作实时流数据。,10,Spark,的组件,MLlib:,Spark有一个包含通用机器学习功能的包，就是MLlib(machine learning lib)。,MLlib包含了分类，聚类，回归，协同过滤算法，还包括模型评估，和数据导入。,它还提供了一些低级的机器学习原语，包括通用梯度下降优化算法。,MLlib提供的上面这些方法，都支持集群上的横向扩展。,11,Spark,的组件,Graphx:,是处理图的库(例如，社交网络图)，并进行图的并行计算。就像Spark Streaming和Spark SQL一样，Graphx也继承了Spark RDD API，同时允许创建有向图。,Graphx提供了各种图的操作，例如subgraph和mapVertices，也包含了常用的图算法，例如PangeRank等。,12,Spark,的组件,Cluster Managers:,Cluster Managers就是集群管理。Spark能够运行在很多cluster managers上面，包括Hadoop YARN，Apache Mesos和Spark自带的单独调度器。,如果你把Spark安装在了裸机上面，单独调度器能够提供简单的方式，让你开始Spark之旅。,如果你已经有了Hadoop Yarn或者Mesos集群，那么，Spark对这些集群管理工具的支持，使你的Spark应用程序能够在这些集群上面运行。,13,Spark,的历史,Spark诞生于2009年，那时候它是，加州大学伯克利分校RAD实验室的一个研究项目，后来到了AMP实验室。,Spark最初是基于Hadoop Mapreduce的，后来发现Mapreduce在迭代式计算和交互式上是低效的。因此Spark进行了改进，引入了内存存储和高容错机制。,关于Spark的研究论文在学术会议上发表，并且在它被创建的2009年不久之后，对于一些特定的工作，Spark比Mapreduce快10-20倍。,2010年3月份Spark开源。,2011年，AMP实验室开始在Spark上面开发高级组件，像Shark(Hive on Spark)，Spark Streaming。,2013年转移到了Apache下，现在已经是顶级项目了。,2014年5月份Spark1.0发布。,14,Spark,运行环境,Spark 是Scala写的,运行在JVM上。所以运行环境是Java6或者以上。,如果想要使用 Python API，需要安装Python 解释器2.6版本或者以上。,目前Spark(1.2.0版本)与Python 3不兼容。,15,Spark,下载,下载地址：spark.apache.org/downloads.html，选择Pre-built for Hadoop 2.4 and later 这个包，点击直接下载，这会下载一个spark-1.2.0-bin-hadoop2.4.tgz的压缩包,搭建Spark不需要Hadoop，如果你有hadoop集群或者hdfs，你可以下载相应的版本。,解压：tar-zxvf spark-1.2.0-bin-hadoop2.4.tgz,16,Spark,目录,README.md,开始Spark之旅的简单介绍。,bin,包含用来和Spark交互的可执行文件，如Spark shell。,core,streaming,python,包含主要组件的源代码。,examples,包含一些有用的单机Spark job。你可以研究和运行这些例子，来学习Spark API。,17,Spark,的,Shells,Spark的shell使你能够处理分布在集群上的数据(这些数据可以是分布在硬盘上或者内存中)。,Spark可以把数据加载到工作节点的内存中，因此，许多分布式处理(甚至是分布式的1T数据的处理)都可以在几秒内完成。,上面的特性，使迭代式计算，实时查询、分析一般能够在shells中完成。Spark提供了Python shells和 Scala shells。,18,Spark,的,Shells,打开Spark的Python Shell：,到Spark目录，Spark的Python Shell也叫做PySpark Shell,bin/pyspark,打开PySpark Shell之后的界面,19,Spark,的,Shells,打开Spark的Scala Shell：,到Spark目录,bin/pysparkbin/spark-shell打开Scala版本的shell,打开之后的界面,20,Spark,的,Shells,例子：,scala val lines=sc.textFile(././testfile/helloSpark),/创建一个叫lines的RDD,lines:org.apache.spark.rdd.RDDString=././testfile/helloSpark MappedRDD1 at textFile at:12,scala lines.count(),/对这个RDD中的行数进行计数,res0:Long=2,scala lines.first(),/文件中的第一行,res1:String=hello spark,修改日志级别：conf/log4j.properties log4j.rootCategory=WARN,console,21,Spark,的核心概念,Driver program：,包含程序的main()方法，RDDs的定义和操作。(在上面的例子中，driver program就是Spark Shell它本身了),它管理很多节点，我们称作executors。,count()操作解释(,每个executor计算文件的一部分，最后合并,)。,22,Spark,的核心概念,SparkContext:,Driver programs 通过一个 SparkContext 对象访问 Spark，SparkContext 对象代表和一个集群的连接。,在Shell中SparkContext 自动创建好了，就是sc，,例子：sc 变量,sc,23,Spark,的核心概念,RDDs：,在Spark中，我们通过分布式集合(distributed collections，也就是RDDs)来进行计算，这些分布式集合，并行的分布在整个集群中。,RDDs 是 Spark分发数据和计算的基础抽象类。,用SparkContext创建RDDs,上面例子中使用sc.textFile()创建了一个RDD，叫lines，它是从我们的本机文本文件中创建的，,这个RDD代表了一个文本文件的每一行,。我们可以在RDD上面进行各种并行化的操作，例如计算数据集中元素的个数或者打印出第一行。,24,Spark,的核心概念,向Spark传递函数：,向Spark传递函数是Spark的一个常用功能，许多Spark API是围绕它展开的。,例子：filtering,scala val lines=sc.textFile(././testfile/helloSpark),lines:spark.RDDString=MappedRDD.,scala val worldLines=lines.filter(line=line.contains(world),pythonLines:spark.RDDString=FilteredRDD.,scala worldLines.collect(),25,Spark,的核心概念,向Spark传递函数：,上面例子中的=语法是 Scala中定义函数的便捷方法。你也可以先定义函数再引用:,例子：,def hasWorld(line:String):Boolean=,line.contains(world),worldLines=lines.filter(hasWorld),像filter 这样的基于函数的操作，也是在集群上并行执行的。,26,Spark,的核心概念,向Spark传递函数：,需要注意的地方：,如果你传递的函数是一个对象的成员，或者包含一个对象中字段的引用(例如self.field)，Spark会把整个对象都发送到工作节点上，这样会比仅仅发送你关心的信息要大很多，而且有时候会带来一些奇怪的问题。,传送信息太多解决方法：我们可以把关心的字段抽取出来，只传递关心的字段。,奇怪问题的避免：序列化包含函数的对象，函数和函数中引用的数据都需要序列化(实现Java的Serializable interface)。,如果Scala中出现NotSerializableException，一般情况下，都是因为没序列化。,27,RDDs,介绍,RDDs介绍,RDDs的创建方法,Scala的基础知识,28,RDDs,介绍,RDDs,Resilient distributed datasets(弹性分布式数据集，简写RDDs)。,一个RDD就是一个,不可改变的分布式集合对象，内部由许多partitions(分片)组成，每个partition都包括一部分数据，,这些partitions可以在集群的不同节点上计算,Partitions是Spark中的并行处理的单元。Spark顺序的，并行的处理partitions。,RDDs 是 Spark的分发数据和计算的基础抽象类，是Spark的核心概念。,RDD可以包含 Python,Java,或者 Scala中的任何数据类型，包括用户自定义的类。,在Spark中，所有的计算都是通过RDDs的创建，转换，操作完成的。,RDD具有lineage graph(血统关系图)。,29,RDDs,的创建方法,Driver program中创建RDDs：,把一个存在的集合传给SparkContexts parallelize()方法。这种方法，一般只适用于学习时。,例子：,val lines=sc.parallelize(List(spark,bigdatastudy),val rdd=sc.parallelize(Array(1,2,2,4),4),.,注意一下RDD的类型,第一个参数是：待并行化处理的集合,第二个参数是：分区个数,30,RDDs,的创建方法,加载外部数据集:,例子：使用textFile()加载,val rddText=sc.textFile(././testfile/helloSpark),val rddHdfs=sc.textFile(hdfs:/some/path.txt),31,Scala,的基础知识,Scala的变量声明,在Scala中创建变量的时候，必须使用val或者var,Val，变量值不可修改，一旦分配不能重新指向别的值,Var，分配后，可以指向类型相同的值。,32,Scala,的基础知识,Scala的变量声明,val lines=sc.textFile(././testfile/helloSpark),lines=sc.textFile(././testfile/helloSpark2),.,:error:reassignment to val,var lines2=sc.textFile(././testfile/helloSpark),lines2=sc.textFile(././testfile/helloSpark2),可以重新声明变量,val lines=sc.textFile(././testfile/helloSpark2),33,Scala,的基础知识,Scala的匿名函数,像Python的lambda 函数,lines.filter(line=line.contains(world),.,我们定义一个匿名函数，接收一个参数line，并使用line这个String类型变量上的contains方法，并且返回结果。,line 的类型不需指定，能够推断出来,34,Scala,的基础知识,Scala程序员就是不喜欢多写代码。,Scala允许我们用下划线_来代表匿名函数中的参数。,lines.filter(_.contains(world),.,35,Scala,的基础知识,类型推断,def hasWorld(line:String):Boolean=line.contains(world),worldLines=lines.filter(hasWorld),Scala中定义函数用def，参数指定类型String，因为后面的contains方法就是用的String中的Contains方法。,函数返回的类型，可以不必指定，因为通过类型推断，能够推出来。,36,Scala,的基础知识,类型推断,指定返回类型,：,返回的类型比较复杂，Scala可能推断不出来。,程序更易读。,37,Transformations,Transformations介绍,逐元素transformations,集合运算,38,Transformations,介绍,Transformations(转换)：,从之前的RDD构建一个,新的RDD,，像map()和 filter()。,39,Transformations,介绍,Transformations的特点：,Transformations返回一个崭新的RDD，,filter()操作返回一个指针，指向一个崭新的RDD，原RDD不受影响，能够在后面重复利用。,40,逐元素,transformations,许多的transformations是逐元素的，也就是每次转变一个元素。,两个最常用的transformations：map()and filter(),map()transformation，接收一个函数，把这个函数应用到RDD的每一个元素，并返一个函数作用后的新的RDD。,filter()transformation，接收一个函数，返回只包含满足filter()函数的元素的新RDD。,输入RDD与输出RDD可以是不同的类型，例如input RDDString，output RDDDouble,41,逐元素,transformations,map(),例子-对RDD中元素求平方,val input=sc.parallelize(List(1,2,3,4),val result=input.map(x=x*x),println(result.collect().mkString(,),42,逐元素,transformations,flatMap(),对每个输入元素，输出多个输出元素。,flat压扁的意思，将RDD中元素压扁后返回一个新的RDD。,例子-flatMap(),把一行字分割成多个元素,val lines=sc.parallelize(List(hello world,hi),val words=lines.flatMap(line=line.split(),words.first()/returns hello,43,逐元素,transformations,flatMap(),44,集合运算,RDDs支持数学集合的计算，例如并集，交集计算。注意：进行计算的RDDs应该是相同类型。,money-monkey,45,集合运算,distinct()是很耗时的操作，因为它需要通过网络，shuffle所有的数据，以保证元素不重复。,一般情况下，我们不用distinct()。,union(other)会包含重复的元素。,intersection(other)求交集。耗时操作，因为需要shuffle,subtract(other)第一个RDD中存在，而不存在与第二个RDD的元素。需要shuffle。使用场景，机器学习中，移除训练集。,46,集合运算,cartesian(other),非常耗时。,使用场景：用户相似性的时候,47,RDD,的,transformations,基本的RDD transformations：RDD 包含 1,2,3,3,函数名,功能,例子,结果,map(),对每个元素应用函数,rdd.map(x=x+1),2,3,4,4,flatMap(),压扁，常用来抽取单词,rdd.flatMap(x=x.to(3),1,2,3,2,3,3,3,filter(),过滤,rdd.filter(x=x!=1),2,3,3,distinct(),去重,rdd.distinct(),1,2,3,sample(withReplacement,fraction,seed),对一个,RDD,取样，是否进行替换,rdd.sample(false,0.5),不确定,48,RDD,的,transformations,两个RDD 的transformations：一个RDD包含 1,2,3，另一个RDD包含 3,4,5,函数名,功能,例子,结果,union(),并集,rdd.union(other),1,2,3,3,4,5,intersection(),交集,rdd.intersection(other),3,subtract(),取存在第一个,RDD,，而不存在第二个,RDD,的元素,(,使用场景，机器学习中，移除训练集,),rdd.subtract(other),1,2,cartesian(),笛卡尔积,rdd.cartesian(other),(1,3),(1,4),(3,5),49,Actions,在RDD上计算出来一个结果，把结果返回给driver program或者保存在外部文件系统上，像count()函数 first()。,count(),返回元素的个数,50,RDD,的,actions,函数名,功能,例子,结果,collect(),返回,RDD,的所有元素,rdd.collect(),1,2,3,3,count(),计数,rdd.count(),4,countByValue(),返回一个,map,，表示唯一元素出现的个数,rdd.countByValue(),(1,1),(2,1),(3,2),take(num),返回几个元素,rdd.take(2),1,2,top(num),返回前几个元素,rdd.top(2),3,3,takeOrdered(num)(ordering),返回基于提供的排序算法的前几个元素,rdd.takeOrdered(2)(myOrdering),3,3,takeSample(withReplacement,num,seed),取样例,rdd.takeSample(false,1),不确定,reduce(func),合并,RDD,中元素,rdd.reduce(x,y)=x+y),9,fold(zero)(func),与,reduce(),相似提供,zero value,rdd.fold(0)(x,y)=x+y),9,aggregate(zeroValue)(seqOp,combOp),与,fold(),相似，返回不同类型,rdd.aggregate(0,0)(x,y)=(x._1+y,x._2+1),(x,y)=(x._1+y._1,x._2+y._2),(9,4),foreach(func),对,RDD,的每个元素作用函数，什么也不返回,rdd.foreach(func),什么也没有,51,Actions,reduce(),最常用的是reduce()，接收一个函数，作用在RDD的两个类型相同的元素上，返回一个,类型相同,的新元素。,最常用的一个函数是加法。,使用reduce()我们可以很简单的实现，RDD中元素的累加，计数，和其它类型的聚集操作。,例子-reduce(),val sum=rdd.reduce(x,y)=x+y),52,Actions,fold(),与reduce()相似，,类型相同,但是，在每个分区的初始化调用的时候，多了个“zero value”,“zero value”的特点，把它应用在你的函数上，不管多少次，都不改变值(例如：+操作的0，*操作的1)。,53,Actions,aggregate(),与fold()相似,类型可以不同,我们提供想要返回的“zero value”类型。,第一个函数，RDD中元素累加(每个节点只累加本地的结果)。,第二个函数，合并累加器(合并每个节点的结果)。,可以使用aggreate()计算RDD的平均值，而不使用map()和fold()结合的方法。,54,Actions,例子-aggregate(),val result=input.aggregate(0,0)(,(x,y)=(x._1+y,x._2+1),(x,y)=(x._1+y._1,x._2+y._2),val avg=result._1/result._2.toDouble,55,Actions,collect(),遍历整个RDD,向driver program返回RDD的内容,一般测试时候使用，可以判断与预测值是否一样,需要单机内存能够容纳下(因为数据要拷贝给driver),大数据的时候，使用saveAsTextFile()action,saveAsSequenceFile()action等。,56,Actions,take(n),返回RDD的n个元素(同时尝试访问最少的partitions)。,返回结果是无序的。,一般测试时候使用,57,Actions,foreach(),计算RDD中的每个元素，但不返回到本地。,可以配合println()友好的打印出数据。,58,Actions,.foreach(println),风格：把函数println当作参数传递给函数foreach,例子-计算bad的个数,errorsRDD=inputRDD.filter(line.contains(error),warningsRDD=inputRDD.filter(line.contains(warning),badLinesRDD=errorsRDD.union(warningsRDD),println(badLinesRDD.count(),badLinesRDD.take(1).foreach(println)/使用take()取前1个数据,59,Actions,top(),排序(根据RDD中数据的比较器),takeSample(withReplacement,num,seed),取样例，是否需要替换值。,countByValue(),返回一个map，表示唯一元素出现的个数,60,