大数据与机器学习第八九章

大数据与机器学习

目录第八章Hive介绍 31.1.1HIVE产生背景 31.1.2组件生态架构 31.1.3HIVE架构 41.1.4部署架构-主备模式 51.1.5处理流程 6第九章Spark介绍 71.1.6Spark产生背景 71.1.7Spark业务架构 91.1.8Spark部署 101.1.9Spark原理 111.1.10Spark-core的RDD 111.1.11产生背景 171.1.12Spark-streaming 181.1.13Spark-graphX图计算 20第八章Hive介绍HIVE产生背景为了使用大数据Hadoop组件，减少MapReduce函数使用的学习成本，Facebook开发了通过使用SQL语言转换为MR函数的工具HIVE。从而可以直接使用MR作业进行数据统计作业，或者直接调用HDFS。组件生态架构HIVE架构Command-lineshell:命令行模式调用显示数据计算结果Thrift与JDBC：支持SQL语句调用显示数据计算结果HueWeb：大数据交互平台，支持Web调用HDFS以图表类型显示数据统计结果。Metastore:元数据存储。主要使用的是关系型数据库（MySQL）。元数据主要是描述数据资源属性信息的数据。如：存储位置、数据字段、数据目录。部署架构-主备模式主备模式，以防HIVE中的MySQL崩溃而丢失数据。处理流程Hive是基于Hadoop的一个数据仓库工具，用来进行数据提取、转化、加载。Hive是将结构化的数据文件映射为数据库表，通过SQL语句底层调用MapReduce进行数据分析。hive十分适合对数据仓库进行统计分析。Hive客户端：由CLI命令行、JDBC(java数据库连接器-JAVA开发)/ODBC（面向对面数据库连接器-C语言开发）、WebGUI网页模式，其中CLI和WebGUI都可以直接命令Driver执行任务提取数据。而JDBC/ODBC方式需要启动ThriftServer(RPC远程调用)，可以允许客户端使用包括Java、C++、Ruby和其他很多种语言，通过编程的方式远程访问Hive

。（即前端服务可以远程调用后端数据）Metastore:因为将数据映射为表，所以需要将元数据（表的列和分区及其属性，表的属性（是否为外部表等），表的数据所在目录等）存入数据库，目前Metastore用的是MySQL关系数据库。将元数据保存在关系数据库中，大大减少了在查询过程中执行语义检查的时间（即数据库会去判断SQL语句是否真正具有相应的含义，比如说sql语句涉及的表，或者字段是否存在等。）Driver：包含解释器、Compiler编译器、Optimizer优化器、Executor执行器解释器：解析sql语句，检查表，字段是否存在。编译器：生成Job任务，当前sql语句会生成几个job优化器：检查sql语句是否可以优化执行器：执行sql,调用MapReduce任务MapReduce任务全过程客户端向JobTracker申请任务，由JobTracker向多个TaskTracker分配任务。再由客户端向YARN的SourceManager申请资源，SourceManager在随机向一台NodeManager发布建立里ApplicationMaster的命令，NodeManager本地建立一个container，然后启动ApplicationMaster，AM再向SM注册和申请资源，SM收到请求后分配资源，并返回给AM。AM收到后，分别到各个NM要求其启动container执行任务，并为任务设置资源。各个NM执行任务，读取本地DataNode的HDFS文件。读取文件过程为，客户端先向NameNode请求访问，请求通过后，由NM将本地最新镜像文件返回给客户端，客户端按照镜像文件的元数据建立各个NM读取通道。第九章Spark介绍Spark产生背景Spark是在MapReduce基础上完善出来的。为了弥补MR的应用不足（实时处理、内存高效、易用提升、迭代处理）。本质上与MR一样，都是大数据处理引擎。以下是与MR相比的优势性提升100倍处理效率：因Spark在内存中处理，比MR处理快100倍。或者在硬盘处理也要快上10倍。提升易用性：Spark提供了80多种high-level操作（ioin、group、sum、count...）更容易建立并行应用，比MR的Map和Reduce的低等级操作要多。完善迭代处理：MapReduce每次操作都要对磁盘进行I\O操作，Spark的通过RDD可以控制中间结果，将结果应用到后续的计算中。最后统计写入HDFS，大大减少了中间结果的存储，降低了数据复制、磁盘I\O和数据序列化开销。完善实时流处理：Streaming流式处理完善实时计算Spark业务架构SparkCore：

实现了Spark的基本功能，包含任务调度、内存管理、错误恢复、与存储系统交互等模块。SparkCore中还包含了对弹性分布式数据集(resilientdistributeddataset，简称RDD)的API定义。

SparkSQL：

是Spark用来操作结构化数据的程序包。通过SparkSQL，我们可以使用SQL或者ApacheHive版本的SQL方言(HQL)来查询数据。SparkSQL支持多种数据源，比如Hive表、Parquet以及JSON等。

SparkStreaming：

是Spark提供的对实时数据进行流式计算的组件。提供了用来操作数据流的API，并且与SparkCore中的RDDAPI高度对应。

SparkMLlib：

提供常见的机器学习(ML)功能的程序库。包括分类、回归、聚类、协同过滤等，还提供了模型评估、数据导入等额外的支持功能。

集群管理器：

Spark设计为可以高效地在一个计算节点到数千个计算节点之间伸缩计算。为了实现这样的要求，同时获得最大灵活性，Spark支持在各种集群管理器(clustermanager)上运行，包括HadoopYARN、ApacheMesos，以及Spark自带的一个简易调度器，叫作独立调度器。Spark部署从物理部署层面上来看，Spark主要分为两种类型的节点，Master节点和Worker节点，Master节点主要运行集群管理器的中心化部分，所承载的作用是分配Application到Worker节点，维护Worker节点，Driver，Application的状态。Worker节点负责具体的业务运行。从Spark程序运行的层面来看，Spark主要分为驱动器节点和执行器节点。Spark原理RDD基本操作RDD的操作分为转化（Transformation）操作和行动（Action）操作。转化操作就是从一个RDD产生一个新的RDD，而行动操作就是进行实际的计算。RDD的操作是惰性的，当RDD执行转化操作的时候，实际计算并没有被执行，只有当RDD执行行动操作时才会促发计算任务提交，从而执行相应的计算操作。1.构建操作Spark里的计算都是通过操作RDD完成的，学习RDD的第一个问题就是如何构建RDD，构建RDD的方式从数据来源角度分为以下两类。从内存里直接读取数据。从文件系统里读取数据，文件系统的种类很多，常见的就是HDFS及本地文件系统。第一类方式是从内存里构造RDD，需要使用makeRDD方法，代码如下所示。valrdd01=sc.makeRDD(List(l,2,3,4,5,6))这个语句创建了一个由“1,2,3,4,5,6”六个元素组成的RDD。第二类方式是通过文件系统构造RDD，代码如下所示valrdd:RDD[String]==sc.textFile(“file:///D:/sparkdata.txt”,1)这里例子使用的是本地文件系统，所以文件路径协议前缀是file://。2.转换操作RDD的转换操作是返回新的RDD的操作。转换出来的RDD是惰性求值的，只有在行动操作中用到这些RDD时才会被计算。许多转换操作都是针对各个元素的，也就是说，这些转换操作每次只会操作RDD中的一个元素，不过并不是所有的转换操作都是这样的。表1描述了常用的RDD转换操作。3.行动操作行动操作用于执行计算并按指定的方式输出结果。行动操作接受RDD，但是返回非RDD，即输出一个值或者结果。在RDD执行过程中，真正的计算发生在行动操作。表2描述了常用的RDD行动操作。RDD血缘关系RDD的最重要的特性之一就是血缘关系（Lineage)，它描述了一个RDD是如何从父RDD计算得来的。如果某个RDD丢失了，则可以根据血缘关系，从父RDD计算得来。图2给出了一个RDD执行过程的实例。系统从输入中逻辑上生成了A和C两个RDD，经过一系列转换操作，逻辑上生成了F这个RDD。Spark记录了RDD之间的生成和依赖关系。当F进行行动操作时，Spark才会根据RDD的依赖关系生成DAG，并从起点开始真正的计算。上述一系列处理称为一个血缘关系（Lineage），即DAG拓扑排序的结果。在血缘关系中，下一代的RDD依赖于上一代的RDD。例如，在图2中，B依赖于A，D依赖于C，而E依赖于B和D。RDD依赖类型根据不同的转换操作，RDD血缘关系的依赖分为窄依赖和宽依赖。窄依赖是指父RDD的每个分区都只被子RDD的一个分区所使用（一对一）。宽依赖是指父RDD的每个分区都被多个子RDD的分区所依赖（一对多）。map、filter、union等操作是窄依赖，而groupByKey、reduceByKey等操作是宽依赖，如图3所示join操作有两种情况，如果join操作中使用的每个Partition仅仅和固定个Partition进行join，则该join操作是窄依赖，其他情况下的join操作是宽依赖。所以可得出一个结论，窄依赖不仅包含一对一的窄依赖，还包含一对固定个数的窄依赖，也就是说，对父RDD依赖的Partition不会随着RDD数据规模的改变而改变。1.窄依赖1）子RDD的每个分区依赖于常数个父分区（即与数据规模无关)。2）输入输出一对一的算子，且结果RDD的分区结构不变，如map、flatMap。3）输入输出一对一的算子，但结果RDD的分区结构发生了变化，如union。4）从输入中选择部分元素的算子，如filter、distinct、subtract、sample。2.宽依赖对单个RDD基于Key进行重组和reduce，如groupByKey、reduceByKey。3）对两个RDD基于Key进行join和重组，如join。Spark的这种依赖关系设计，使其具有了天生的容错性，大大加快了Spark的执行速度。RDD通过血缘关系记住了它是如何从其他RDD中演变过来的。当这个RDD的部分分区数据丢失时，它可以通过血缘关系获取足够的信息来重新运算和恢复丢失的数据分区，从而带来性能的提升。相对而言，窄依赖的失败恢复更为高效，它只需要根据父RDD分区重新计算丢失的分区即可，而不需要重新计算父RDD的所有分区。而对于宽依赖来讲，单个结点失效，即使只是RDD的一个分区失效，也需要重新计算父RDD的所有分区，开销较大。宽依赖操作就像是将父RDD中所有分区的记录进行了“洗牌”，数据被打散，然后在子RDD中进行重组。阶段划分DAGDAG（DirectedAcyclicGraph有向无环图）又称“血统”Lineage。用户提交的计算任务是一个由RDD构成的DAG，如果RDD的转换是宽依赖，那么这个宽依赖转换就将这个DAG分为了不同的阶段（Stage)。由于宽依赖会带来“洗牌”，所以不同的Stage是不能并行计算的，后面Stage的RDD的计算需要等待前面Stage的RDD的所有分区全部计算完毕以后才能进行。这点就类似于在MapReduce中，Reduce阶段的计算必须等待所有Map任务完成后才能开始一样。在对Job中的所有操作划分Stage时，一般会按照倒序进行，即从Action开始，遇到窄依赖操作，则划分到同一个执行阶段，遇到宽依赖操作，则划分一个新的执行阶段。后面的Stage需要等待所有的前面的Stage执行完之后才可以执行，这样Stage之间根据依赖关系就构成了一个大粒度的DAG。下面通过图4详细解释一下阶段划分。假设从HDFS中读入数据生成3个不同的RDD(A、C和E)，通过一系列转换操作后得到新的RDD(G)，并把结果保存到HDFS中。可以看到这幅DAG中只有join操作是一个宽依赖，Spark会以此为边界将其前后划分成不同的阶段。同时可以注意到，在Stage2中，从map到union都是窄依赖，这两步操作可以形成一个流水线操作，通过map操作生成的分区可以不用等待整个RDD计算结束，而是继续进行union操作，这样大大提高了计算的效率。把一个DAG图划分成多个Stage以后，每个Stage都代表了一组由关联的、相互之间没有宽依赖关系的任务组成的任务集合。在运行的时候，Spark会把每个任务集合提交给任务调度器进行处理。产生背景ApacheFlink是由Apache软件基金会开发的开源流处理框架，其核心是用Java和Scala编写的分布式流数据处理引擎。Flink以数据并行和流水线方式执行任意流数据程序，Flink的流水线运行时系统可以执行批处理和流处理程序。此外，Flink的运行时本身也支持迭代算法的执行。Flink提供高吞吐量、低延迟的流数据处理引擎以及对事件-时间处理和状态管理的支持。Flink应用程序在发生机器故障时具有容错能力，并且支持exactly-once语义。（严格一次）Spark-streamingStreaming：是一种数据传送技术，它把客户机收到的数据变成一个稳定连续的流，源源不断地送出，使用户听到的声音或看到的图象十分平稳，而且用户在整个文件送完之前就可以开始在屏幕上浏览文件。常见的流式计算框架ApacheStormSparkStreamingApacheSamzaSparkStreaming是SparkCore的扩展应用，它具有可扩展，高吞吐量，对于流数据的可容错性等特点。可以监控来自Kafka,Flume,HDFS。Kinesis,Twitter，ZeroMQ或者Scoket套接字的数据通过复杂的算法以及一系列的计算分析数据，并且可以将分析结果存入到HDFS文件系统，数据库以及前端页面中。SparkStreaming有以下特点：高可扩展性，可以运行在上百台机器上（Scalestohundredsofnodes）低延迟，可以在秒级别上对数据进行处理（Achieveslowlatency）高可容错性（Efficientlyrecoverfromfailures）能够集成并行计算程序，比如SparkCore（Integrateswithbatchandinteractiveprocessing）工作原理对于SparkCore它的核心就是RDD，对于SparkStreaming来说，它的核心是DStream，DStream类似于RDD，它实质上一系列的RDD的集合，DStream可以按照秒数将数据流进行批量的划分。首先从接收到流数据之后，将其划分为多个batch，然后提交给Spark集群进行计算，最后将结果批量输出到HDFS或者数据库以及前端页面展示等等。可以参考下面这幅图来帮助理解：DStream是一系列连续的RDD，它是建立在Spark之上的不可变的，分布式数据集，在DStream中的每一个RDD包含着一定时间间隔的数据，如下图所示：SparkCore在初始化时会生成一个SparkContext对象来对数据进行后续的处理，相对应的SparkStreaming会创建一个StreamingContext，它的底层是SparkContext，也就是说它会将任务提交给SparkContext来执行，这也很好的解释了DStream是一系列的RDD。当启动SparkStreaming应用的时候，首先会在一个节点的Executor上启动一个Receiver接受者，然后当从数据源写入数据的时候会被Receiver接收，接收到数据之后Receiv