Spark快速大数据分析

Spark快速大数据分析广东航信爱信诺科技公司——研发部2019-04-18Spark简介1、Spark的定位：是一个用来实现快速而通用的集群计算平台。2、Spark与Hadoop的联系：Spark扩展了mapreduce计算模型，且支持更多的计算模式，包括交互式查询和流处理。3、Spark的主要特点：能够在内存中进行计算，因而更快。即便是在磁盘上进行复杂的计算，Spark依然比mapreduce更高效。Spark简介Spark主要包含了如下图所示的组件：1、SparkCore:实现了Spark的基本功能，包含任务调度、内存管理、错误恢复与存储系统交互等模块，还包含了对弹性分布式数据集(ResilientDistributedDataset)的API定义。2、SparkSQL:是Spark操作结构化数据的程序包，通过SparkSQL可以使用SQL或者ApacheHive版本的SQL方言(HQL)来查询数据。3、SparkStreaming:是Spark提供的对实时数据进行流式计算的组件，如生产环境中的网页服务器日志。4、MLlib:提供机器学习功能的程序库。5、GraphX:操作图的程序库，可以进行并行的图计算。6、集群管理器：为了实现在多个计算节点上高效得伸缩计算，Spark支持在各种集群管理器上运行，包括HadoopYARN、ApacheMesos以及自带的简易调度器，独立调度器，第七章部分会详细探讨管理器的差异以及如何选择合适的集群管理器。核心概念与基本操作弹性分布式数据集(ResilientDistributedDataset,简称RDD):RDD的核心特性是分布式与不可变。Spark会自动将RDD中的数据分发到集群上，并将操作并行化执行。每一个Spark应用都由一个驱动器程序(driverprogram)来发起集群上的各种并行操作，驱动器程序通过一个SparkContext对象来访问Spark，使用pythonshell启动时会自动创建一个SparkContext对象，默认变量名为sc。核心概念与基本操作Spark中对数据的所有操作不外乎：1、创建RDD2、转化已有RDD，即转化操作(transformation):由一个RDD生成一个新的RDD3、调用RDD操作进行求值，即行动操作(action):会对一个RDD计算出一个结果创建RDD的方式:1、通过已有集合生成，用于原型开发和测试lines=sc.parallelize([“hello,“helloworld])2、从外部存储中读取数据lines=sc.textFile(/path/to/helloworld.txt)核心概念与基本操作转化操作举例:1、filter操作筛选出RDD1中满足条件的元素构成新的RDD，以下意为选出大于5的元素构成新RDDnewRDD=oldRDD.filter(lambdax:x5)2、map操作对RDD1中的每个元素进行函数操作后，将结果构成新的RDD，以下意为对RDD中的每个元素进行平方构成新RDDnewRDD=oldRDD.map(lambdax:x**2)3、flatMap操作和map操作类似，但是如果操作后的结果为一个列表，则取出列表中的元素构成新RDD，而非将列表构成新RDDnewRDD=oldRDD.flatMap(lambdax:x.split())核心概念与基本操作其中map()和flatMap()操作的区别如下图所示：核心概念与基本操作行动操作举例：1、count操作用于RDD计数print(Inputhas+errorRDD.count()+errorlines)2、take操作用于取出RDD默认排序前n个元素forlineinerrorRDD.take(5):print(line)核心概念与基本操作操作中可能会设计传递各种函数，spark传递函数主要有以下几种方式：1、传递函数较简单，可通过匿名函数wordRDD=lineRDD.filter(lambdax:'word'inx)2、传递顶层或局部定义的函数defcontainsError(s):returnerrorinserrorRDD=lineRDD.filter(containsError)核心概念与基本操作Spark具有惰性计算的特性：意思是只有在一个RDD第一次执行行动操作时，才会真正计算。举例来说，RDD2由RDD1生成，但是只有在RDD2执行行动操作时，才会开始计算RDD2，而不是在RDD1生成RDD2时就进行计算。但Spark会使用谱系图(lineagegragh)来记录不同RDD之间的依赖关系，如下图默认情况下，Spark的RDD会在每次对他们进行行动操作时重新计算，如果重用同一个RDD，可以使用RDD.persist()让Spark把这个RDD缓存下来。举例来说，有一个原始RDD1，由RDD1生成一个RDD2，再由RDD2生成了RDD3，这时如果需要用RDD3进行两个行动操作，默认情况下，系统两次都会从RDD1开始重新计算RDD3，再进行行动操作。如果在RDD3生成后，使用RDD3.persist(),则两次行动操作就会忽略RDD1到RDD3的计算过程，从而省略了1次RDD3的生成过程。类似于闯关游戏中存储进度，下次从存储位置继续玩起的动作。核心概念与基本操作尽管RDD本身不是严格意义上的集合，因为常常有重复的数据，但是仍然支持一些集合操作：1、union操作，生成一个包含两个RDD所有元素的新RDD，类似于并集，若元素重复不会去重newRDD=RDD1.union(RDD2)2、intersection操作，生成两个RDD中共同拥有的元素，类似于交集，会对重复元素去重newRDD=RDD1.intersection(RDD2)核心概念与基本操作3、subtract操作，生成一个元素只存在于RDD1，不存在于RDD2的新RDD，subtract会保留newRDD中的重复元素newRDD=RDD1.subtract(RDD2)4、distinct操作，生成一个去重后的新RDDnewRDD=RDD.distinct()5、cartesian操作，返回RDD1和RDD2的笛卡尔积，大规模的笛卡尔积开销巨大newRDD=RDD1.cartesian()核心概念与基本操作两个RDD的笛卡尔积如下图：