Spark各个知识点总结省公共课一等奖全国赛课获奖课件.pptx

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