Spark各个知识点总结.ppt

Spark各个知识点总结.ppt

《Spark各个知识点总结.ppt》由会员分享，可在线阅读，更多相关《Spark各个知识点总结.ppt（60页珍藏版）》请在冰点文库上搜索。

Spark简介,Spark是什么,Spark是一个快速且通用的集群计算平台。

集群计算把一台电脑无法解决的问题，放到多台电脑组成的集群上进行解决，这就是集群计算。

Spark的特点,Spark是快速的很多任务能够秒级完成，对于一些特定的工作，Spark比Mapreduce快10-20倍。

Spark扩充了流行的Mapreduce计算模型，使Spark更高效地支持更多类型的计算，包括交互式查询，和流处理。

速度快的另一个主要原因就是，能够在内存中计算。

Spark的特点,Spark是通用的Spark的设计，容纳了之前很多独立的，分布式系统所拥有的功能。

独立的分布式系统包括：

批处理，迭代式计算，交互查询和流处理等。

并且，由之前需要维护不同的集群，到现在只需要维护一个Spark集群。

Spark的特点,Spark是高度开放的Spark提供了Python，Java，Scala，SQL的API和丰富的内置库。

同时，Spark和其它的大数据工具整合的很好。

尤其，Spark能够运行在Hadoop集群上面，能够访问Hadoop数据。

Spark的组件,Spark包括多个紧密集成的组件。

Spark的组件,紧密集成的优点：

如果Spark底层优化了，那么基于Spark底层的组件，也得到了相应的优化。

例如，Spark底层增加了一个优化算法，那么Spark的SQL和机器学习包也会自动的优化。

紧密集成，节省了各个组件组合使用时的部署，测试等时间。

当向Spark增加新的组件时，其它的组件，可以立刻享用新组件的功能。

无缝连接不同的处理模型。

Spark的组件,SparkCore：

包含Spark的基本功能，包含任务调度，内存管理，容错机制等。

SparkCore内部定义了RDDs（resilientdistributeddatasets，弹性分布式数据集）。

RDDs代表横跨很多工作节点的数据集合，RDDs可以被并行的处理。

SparkCore提供了很多APIs来创建和操作这些集合（RDDs）。

Spark的组件,SparkSQL：

是Spark处理结构化数据的库。

它支持通过SQL查询数据，就像HQL（HiveSQL）一样，并且支持很多数据源，像Hive表，JSON等。

SparkSQL是在Spark1.0版本中新加的。

Shark是一种较老的基于Spark的SQL项目，它是基于Hive修改的，它现在已经被SparkSQL替代了。

Spark的组件,SparkStreaming:

是实时数据流处理组件，类似Storm。

SparkStreaming提供了API来操作实时流数据。

Spark的组件,MLlib:

Spark有一个包含通用机器学习功能的包，就是MLlib（machinelearninglib）。

MLlib包含了分类，聚类，回归，协同过滤算法，还包括模型评估，和数据导入。

它还提供了一些低级的机器学习原语，包括通用梯度下降优化算法。

MLlib提供的上面这些方法，都支持集群上的横向扩展。

Spark的组件,Graphx:

是处理图的库（例如，社交网络图），并进行图的并行计算。

就像SparkStreaming和SparkSQL一样，Graphx也继承了SparkRDDAPI，同时允许创建有向图。

Graphx提供了各种图的操作，例如subgraph和mapVertices，也包含了常用的图算法，例如PangeRank等。

Spark的组件,ClusterManagers:

ClusterManagers就是集群管理。

Spark能够运行在很多clustermanagers上面，包括HadoopYARN，ApacheMesos和Spark自带的单独调度器。

如果你把Spark安装在了裸机上面，单独调度器能够提供简单的方式，让你开始Spark之旅。

如果你已经有了HadoopYarn或者Mesos集群，那么，Spark对这些集群管理工具的支持，使你的Spark应用程序能够在这些集群上面运行。

Spark的历史,Spark诞生于2009年，那时候它是，加州大学伯克利分校RAD实验室的一个研究项目，后来到了AMP实验室。

Spark最初是基于HadoopMapreduce的，后来发现Mapreduce在迭代式计算和交互式上是低效的。

因此Spark进行了改进，引入了内存存储和高容错机制。

关于Spark的研究论文在学术会议上发表，并且在它被创建的2009年不久之后，对于一些特定的工作，Spark比Mapreduce快10-20倍。

2010年3月份Spark开源。

2011年，AMP实验室开始在Spark上面开发高级组件，像Shark（HiveonSpark），SparkStreaming。

2013年转移到了Apache下，现在已经是顶级项目了。

2014年5月份Spark1.0发布。

Spark运行环境,Spark是Scala写的,运行在JVM上。

所以运行环境是Java6或者以上。

如果想要使用PythonAPI，需要安装Python解释器2.6版本或者以上。

目前Spark（1.2.0版本）与Python3不兼容。

Spark下载,下载地址：

http:

/spark.apache.org/downloads.html，选择Pre-builtforHadoop2.4andlater这个包，点击直接下载，这会下载一个spark-1.2.0-bin-hadoop2.4.tgz的压缩包搭建Spark不需要Hadoop，如果你有hadoop集群或者hdfs，你可以下载相应的版本。

解压：

tar-zxvfspark-1.2.0-bin-hadoop2.4.tgz,Spark目录,README.md开始Spark之旅的简单介绍。

bin包含用来和Spark交互的可执行文件，如Sparkshell。

core,streaming,python,包含主要组件的源代码。

examples包含一些有用的单机Sparkjob。

你可以研究和运行这些例子，来学习SparkAPI。

Spark的Shells,Spark的shell使你能够处理分布在集群上的数据（这些数据可以是分布在硬盘上或者内存中）。

Spark可以把数据加载到工作节点的内存中，因此，许多分布式处理（甚至是分布式的1T数据的处理）都可以在几秒内完成。

上面的特性，使迭代式计算，实时查询、分析一般能够在shells中完成。

Spark提供了Pythonshells和Scalashells。

Spark的Shells,打开Spark的PythonShell：

到Spark目录，Spark的PythonShell也叫做PySparkShellbin/pyspark打开PySparkShell之后的界面,Spark的Shells,打开Spark的ScalaShell：

到Spark目录bin/pysparkbin/spark-shell打开Scala版本的shell打开之后的界面,Spark的Shells,例子：

scalavallines=sc.textFile（././testfile/helloSpark）/创建一个叫lines的RDDlines:

org.apache.spark.rdd.RDDString=././testfile/helloSparkMappedRDD1attextFileat:

12scalalines.count（）/对这个RDD中的行数进行计数res0:

Long=2scalalines.first（）/文件中的第一行res1:

String=hellospark修改日志级别：

conf/log4j.propertieslog4j.rootCategory=WARN,console,Spark的核心概念,Driverprogram：

包含程序的main（）方法，RDDs的定义和操作。

（在上面的例子中，driverprogram就是SparkShell它本身了）它管理很多节点，我们称作executors。

count（）操作解释（每个executor计算文件的一部分，最后合并）。

Spark的核心概念,SparkContext:

Driverprograms通过一个SparkContext对象访问Spark，SparkContext对象代表和一个集群的连接。

在Shell中SparkContext自动创建好了，就是sc，例子：

sc变量sc,Spark的核心概念,RDDs：

在Spark中，我们通过分布式集合（distributedcollections，也就是RDDs）来进行计算，这些分布式集合，并行的分布在整个集群中。

RDDs是Spark分发数据和计算的基础抽象类。

用SparkContext创建RDDs上面例子中使用sc.textFile（）创建了一个RDD，叫lines，它是从我们的本机文本文件中创建的，这个RDD代表了一个文本文件的每一行。

我们可以在RDD上面进行各种并行化的操作，例如计算数据集中元素的个数或者打印出第一行。

Spark的核心概念,向Spark传递函数：

向Spark传递函数是Spark的一个常用功能，许多SparkAPI是围绕它展开的。

例子：

filteringscalavallines=sc.textFile（././testfile/helloSpark）lines:

spark.RDDString=MappedRDD.scalavalworldLines=lines.filter（line=line.contains（world）pythonLines:

spark.RDDString=FilteredRDD.scalaworldLines.collect（）,Spark的核心概念,向Spark传递函数：

上面例子中的=语法是Scala中定义函数的便捷方法。

你也可以先定义函数再引用:

例子：

defhasWorld（line:

String）:

Boolean=line.contains（world）worldLines=lines.filter（hasWorld）像filter这样的基于函数的操作，也是在集群上并行执行的。

Spark的核心概念,向Spark传递函数：

需要注意的地方：

如果你传递的函数是一个对象的成员，或者包含一个对象中字段的引用（例如self.field），Spark会把整个对象都发送到工作节点上，这样会比仅仅发送你关心的信息要大很多，而且有时候会带来一些奇怪的问题。

传送信息太多解决方法：

我们可以把关心的字段抽取出来，只传递关心的字段。

奇怪问题的避免：

序列化包含函数的对象，函数和函数中引用的数据都需要序列化（实现Java的Serializableinterface）。

如果Scala中出现NotSerializableException，一般情况下，都是因为没序列化。

RDDs介绍,RDDs介绍RDDs的创建方法Scala的基础知识,RDDs介绍,RDDsResilientdistributeddatasets（弹性分布式数据集，简写RDDs）。

一个RDD就是一个不可改变的分布式集合对象，内部由许多partitions（分片）组成，每个partition都包括一部分数据，这些partitions可以在集群的不同节点上计算Partitions是Spark中的并行处理的单元。

Spark顺序的，并行的处理partitions。

RDDs是Spark的分发数据和计算的基础抽象类，是Spark的核心概念。

RDD可以包含Python,Java,或者Scala中的任何数据类型，包括用户自定义的类。

在Spark中，所有的计算都是通过RDDs的创建，转换，操作完成的。

RDD具有lineagegraph（血统关系图）。

RDDs的创建方法,Driverprogram中创建RDDs：

把一个存在的集合传给SparkContextsparallelize（）方法。

这种方法，一般只适用于学习时。

例子：

vallines=sc.parallelize（List（spark,bigdatastudy）valrdd=sc.parallelize（Array（1,2,2,4）,4）.注意一下RDD的类型第一个参数是：

待并行化处理的集合第二个参数是：

分区个数,RDDs的创建方法,加载外部数据集:

例子：

使用textFile（）加载valrddText=sc.textFile（././testfile/helloSpark）valrddHdfs=sc.textFile（hdfs:

/some/path.txt）,Scala的基础知识,Scala的变量声明在Scala中创建变量的时候，必须使用val或者varVal，变量值不可修改，一旦分配不能重新指向别的值Var，分配后，可以指向类型相同的值。

Scala的基础知识,Scala的变量声明vallines=sc.textFile（././testfile/helloSpark）lines=sc.textFile（././testfile/helloSpark2）.:

error:

reassignmenttovalvarlines2=sc.textFile（././testfile/helloSpark）lines2=sc.textFile（././testfile/helloSpark2）可以重新声明变量vallines=sc.textFile（././testfile/helloSpark2）,Scala的基础知识,Scala的匿名函数像Python的lambda函数lines.filter（line=line.contains（world）.我们定义一个匿名函数，接收一个参数line，并使用line这个String类型变量上的contains方法，并且返回结果。

line的类型不需指定，能够推断出来,Scala的基础知识,Scala程序员就是不喜欢多写代码。

Scala允许我们用下划线_来代表匿名函数中的参数。

lines.filter（_.contains（world）.,Scala的基础知识,类型推断defhasWorld（line:

String）:

Boolean=line.contains（world）worldLines=lines.filter（hasWorld）Scala中定义函数用def，参数指定类型String，因为后面的contains方法就是用的String中的Contains方法。

函数返回的类型，可以不必指定，因为通过类型推断，能够推出来。

Scala的基础知识,类型推断指定返回类型：

返回的类型比较复杂，Scala可能推断不出来。

程序更易读。

Transformations,Transformations介绍逐元素transformations集合运算,Transformations介绍,Transformations（转换）：

从之前的RDD构建一个新的RDD，像map（）和filter（）。

Transformations介绍,Transformations的特点：

Transformations返回一个崭新的RDD，filter（）操作返回一个指针，指向一个崭新的RDD，原RDD不受影响，能够在后面重复利用。

逐元素transformations,许多的transformations是逐元素的，也就是每次转变一个元素。

两个最常用的transformations：

map（）andfilter（）map（）transformation，接收一个函数，把这个函数应用到RDD的每一个元素，并返一个函数作用后的新的RDD。

filter（）transformation，接收一个函数，返回只包含满足filter（）函数的元素的新RDD。

输入RDD与输出RDD可以是不同的类型，例如inputRDDString，outputRDDDouble,逐元素transformations,map（）例子-对RDD中元素求平方valinput=sc.parallelize（List（1,2,3,4）valresult=input.map（x=x*x）println（result.collect（）.mkString（,）,逐元素transformations,flatMap（）对每个输入元素，输出多个输出元素。

flat压扁的意思，将RDD中元素压扁后返回一个新的RDD。

例子-flatMap（）,把一行字分割成多个元素vallines=sc.parallelize（List（helloworld,hi）valwords=lines.flatMap（line=line.split（）words.first（）/returnshello,逐元素transformations,flatMap（）,集合运算,RDDs支持数学集合的计算，例如并集，交集计算。

注意：

进行计算的RDDs应该是相同类型。

money-monkey,集合运算,distinct（）是很耗时的操作，因为它需要通过网络，shuffle所有的数据，以保证元素不重复。

一般情况下，我们不用distinct（）。

union（other）会包含重复的元素。

intersection（other）求交集。

耗时操作，因为需要shufflesubtract（other）第一个RDD中存在，而不存在与第二个RDD的元素。

需要shuffle。

使用场景，机器学习中，移除训练集。

集合运算,cartesian（other）非常耗时。

使用场景：

用户相似性的时候,RDD的transformations,基本的RDDtransformations：

RDD包含1,2,3,3,RDD的transformations,两个RDD的transformations：

一个RDD包含1,2,3，另一个RDD包含3,4,5,Actions,在RDD上计算出来一个结果，把结果返回给driverprogram或者保存在外部文件系统上，像count（）函数first（）。

count（）返回元素的个数,RDD的actions,Actions,reduce（）最常用的是reduce（），接收一个函数，作用在RDD的两个类型相同的元素上，返回一个类型相同的新元素。

最常用的一个函数是加法。

使用reduce（）我们可以很简单的实现，RDD中元素的累加，计数，和其它类型的聚集操作。

例子-reduce（）valsum=rdd.reduce（x,y）=x+y）,Actions,fold（）与reduce（）相似，类型相同但是，在每个分区的初始化调用的时候，多了个“zerovalue”“zerovalue”的特点，把它应用在你的函数上，不管多少次，都不改变值（例如：

+操作的0，*操作的1）。

Actions,aggregate（）与fold（）相似类型可以不同我们提供想要返回的“zerovalue”类型。

第一个函数，RDD中元素累加（每个节点只累加本地的结果）。

第二个函数，合并累加器（合并每个节点的结果）。

可以使用aggreate（）计算RDD的平均值，而不使用map（）和fold（）结合的方法。

Actions,例子-aggregate（）valresult=input.aggregate（0,0）（x,y）=（x._1+y,x._2+1）,（x,y）=（x._1+y._1,x._2+y._2）valavg=result._1/result._2.toDouble,Actions,collect（）遍历整个RDD,向driverprogram返回RDD的内容一般测试时候使用，可以判断与预测值是否一样需要单机内存能够容纳下（因为数据要拷贝给driver）大数据的时候，使用saveAsTextFile（）action,saveAsSequenceFile（）action等。

Actions,take（n）返回RDD的n个元素（同时尝试访问最少的partitions）。

返回结果是无序的。

一般测试时候使用,Actions,foreach（）计算RDD中的每个元素，但不返回到本地。

可以配合println（）友好的打印出数据。

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个数据,Actions,top（）排序（根据RDD中数据的比较器）takeSample（withReplacement,num,seed）取样例，是否需要替换值。

countByValue（）返回一个map，表示唯一元素出现的个数,