5、spark实时计算

Spark实时计算什么是Spark•Spark是UCBerkeleyAMPlab所开源的类HadoopMapReduce的通用的并行计算框架，Spark基于mapreduce算法实现的分布式计算，拥有HadoopMapReduce所具有的优点；但不同于MapReduce的是Job中间输出和结果可以保存在内存中，从而不再需要读写HDFS，因此Spark能更好地适用于数据挖掘与机器学习等需要迭代的mapreduce的算法。Spark发展什么是SparkStreaming•SparkStreaming类似于ApacheStorm，用于流式数据的处理。根据其官方文档介绍，SparkStreaming有高吞吐量和容错能力强这两个特点。SparkStreaming支持的数据输入源很多，例如：Kafka、Flume、Twitter、ZeroMQ和简单的TCP套接字等等。数据输入后可以用Spark的高度抽象原语如：map、reduce、join、window等进行运算。而结果也能保存在很多地方，如HDFS，数据库等。另外SparkStreaming也能和MLlib（机器学习）以及Graphx完美融合。•在SparkStreaming中，处理数据的单位是一批而不是单条，而数据采集却是逐条进行的，因此SparkStreaming系统需要设置间隔使得数据汇总到一定的量后再一并操作，这个间隔就是批处理间隔。批处理间隔是SparkStreaming的核心概念和关键参数，它决定了SparkStreaming提交作业的频率和数据处理的延迟，同时也影响着数据处理的吞吐量和性能。Spark生态圈（1）Spark生态圈（2）•SparkCore–用于通用分布式数据处理的引擎。它不不依赖于任何其他组件，可以运行在任何商用服务器集群上；•SparkSql–运行在Spark上的SQL查询语句，支持一系列SQL函数和HiveQL。但是还不是很成熟，所以不要在生产系统中使用；而HiveQL集成了需要的hive元数据和Hive相关的jar包；•SparkStreaming–基于spark的微批处理引擎，支持各种各样数据源的导入。唯一依赖的是SparkCore引擎；•MLib–构建在spark之上的机器学习库，支持一系列数据挖掘算法；•GraphX-是一个分布式图处理框架；Spark、Hadoop执行说明Spark会代替Hadoop吗•日常工作常用到的需求场景批处理、实时计算、机器学习•Hadoop适合于离线的批量数据处理适用于对实时性要求极低的场景•Storm适合于实时流数据处理，实时性方面做得极好•Spark是内存分布式计算框架，试图吞并Hadoop的Map-Reduce批处理框架和Storm的流处理框架，但是Spark已经做得很不错了，批处理方面性能优于Map-Reduce，但是流处理目前还是弱于Storm，产品仍在改进之中参考：运行方式•单机（开发、测试）部署在单机上时，既可以用本地模式运行，也可以用伪分布模式运行；SparkOnLocal（本地运行）SparkOnLocalCluster（本地伪分布）•集群（测试、生产）YarnClient（测试）：Driver在客户端本地运行，这种模式可以使得SparkApplication和客户端进行交互YarnCluster模式（生产）：Driver在集群运行，客户端提交完作业就可以撤离Spark核心概念-1•(RDD)弹性分布数据集Rdd是spark的灵魂，中文翻译弹性分布式数据集，一个rdd它表示已被分区，不可变的并能够被并行操作的数据集合，不同的数据集格式对应不同的RDD实现。rdd内部可以有许多分区(partitions)，每个分区又拥有大量的记录(records)；RDD是Spark的最基本抽象,是对分布式内存的抽象使用，实现了以操作本地集合的方式来操作分布式数据集的抽象实现；•RDD特点•它是在集群节点上的不可变的、已分区的集合对象；•通过并行转换的方式来创建如（map,filter,join,etc）；•失败自动重建；•可以控制存储级别（内存、磁盘等）来进行重用；•必须是可序列化的；Spark核心概念-2•rdd的五个特征：•Dependencies:建立RDD的依赖关系，主要rdd之间是宽窄依赖的关系，具有窄依赖关系的rdd可以在同一个stage中进行计算。•partition：一个rdd会有若干个分区，分区的大小决定了对这个rdd计算的粒度，每个rdd的分区的计算都在一个单独的任务中进行。•preferedlocations:按照“移动数据不如移动计算”原则，在spark进行任务调度的时候，优先将任务分配到数据块存储的位置•Compute：spark中的计算都是以分区为基本单位的，compute函数只是对迭代器进行复合，并不保存单次计算的结果。•Partitioner：只存在于（K,V）类型的rdd中，非（K,V）类型的partitioner的值就是None。Spark核心概念-RDD如何保障数据处理效率？•分区和持久化：RDD提供了两方面的特性persistence和patitioning，用户可以通过persist与patitionBy函数来控制RDD的这两个方面。RDD的分区特性与并行计算能力(RDD定义了parallerize函数)，使得Spark可以更好地利用可伸缩的硬件资源。若将分区与持久化二者结合起来，就能更加高效地处理海量数据。•内存存储：RDD本质上是一个内存数据集，在访问RDD时，指针只会指向与操作相关的部分。例如存在一个面向列的数据结构，其中一个实现为Int的数组，另一个实现为Float的数组。如果只需要访问Int字段，RDD的指针可以只访问Int数组，避免了对整个数据结构的扫描。•惰性计算：RDD将操作分为两类：transformation与action。无论执行了多少次transformation操作，RDD都不会真正执行运算，只有当action操作被执行时，运算才会触发。而在RDD的内部实现机制中，底层接口则是基于迭代器的，从而使得数据访问变得更高效，也避免了大量中间结果对内存的消耗。Spark核心概念-RDD对容错的支持（一）•目前分布式数据集的容错性有两种方式：数据检查点、记录数据的更新。•数据检查点操作成本很高，需要通过数据中心的网络连接在机器之间复制庞大的数据集；Spark选择记录更新的方式。但是，如果更新粒度太细太多，那么记录更新成本也不低。因此，RDD只支持粗粒度转换；•RDD的容错机制又称“血统(Lineage)”容错；Lineage本质上很类似于数据库中的重做日志（RedoLog），只不过这个重做日志粒度很大；•RDD的Lineage记录的是粗颗粒度的特定数据Transformation操作（如filter、map、join等）行为。当这个RDD的部分分区数据丢失时，它可以通过Lineage获取足够的信息来重新运算和恢复丢失的数据分区。因为这种粗颗粒的数据模型，限制了Spark的运用场合，所以Spark并不适用于所有高性能要求的场景，但同时相比细颗粒度的数据模型，也带来了性能的提升。Spark核心概念-RDD对容错的支持（二）•在容错机制中，如果一个节点死机了，而且运算窄依赖，则只要把丢失的父RDD分区重算即可，不依赖于其他节点。而宽依赖需要父RDD的所有分区都存在，重算就很昂贵了。在宽依赖情况下，丢失一个子RDD分区重算的每个父RDD的每个分区的所有数据并不是都给丢失的子RDD分区用的，会有一部分数据相当于对应的是未丢失的子RDD分区中需要的数据，这样就会产生冗余计算开销，这也是宽依赖开销更大的原因。因此如果使用Checkpoint算子来做检查点，不仅要考虑Lineage是否足够长，也要考虑是否有宽依赖，对宽依赖加Checkpoint是最物有所值的。•下面请看建议使用checkpoint：①DAG中的Lineage过长，如果重算，则开销太大（如在PageRank中）。②在宽依赖上做Checkpoint获得的收益更大。③在SparkStreaming中，针对数据进行update操作，或者调用Streaming提供的window操作时，就需要恢复执行过程的中间状态。Spark编程–TransForm(转换操作)Action(行动操作)参考：编程–示例（worldcount）查看代码：A_WordCount.scalaSpark编程–示例（计算平均年龄）查看代码：B_AvgAgeCalculator.scala第一步：计算总共有多少人第二步：计算所有人的年龄和第三步：使用年龄和/人数Spark编程–示例（分性别身高统计）查看代码：C_UserInfoCalculator.scala第一步：把男、女分成两组数据，数据只保留身高信息第二步：统计每组数据的条数第三步：取得每组数据的最大值、最小值SparkStreaming介绍（1）SparkStreaming是建立在Spark上的实时计算框架，通过它提供的丰富API、基于内存的高速执行引擎，用户可以结合流式、批处理和交互查询应用。特点：1、Spark低延迟执行引擎，可以用于实时数据处理2、相比于storm（基于记录），RDD数据更容易做容错3、可以使用Kafka、Flume、ZeroMQ等作为数据源进行对接4、小批量处理的方式使得它可以同时兼容批量和实时数据处理的逻辑和算法。方便一些需要历史数据和实时数据联合分析的特定应用场景结合SparkStreaming在内部的处理机制是，接收实时流的数据，并根据一定的时间间隔拆分成一批批的数据，然后通过SparkEngine处理这些批数据，最终得到处理后的一批批结果数据。SparkStreaming介绍（2）SparkStreaming的基本原理是将输入数据流以时间片（秒级）为单位进行拆分，然后以类似批处理的方式处理每个时间片数据。1、SparkStreaming把实时输入数据流以时间片Δt（如1秒）为单位切分成块2、SparkStreaming会把每块数据作为一个RDD，并使用RDD操作处理每一小块数据3、每个块都会生成一个SparkJob处理，最终结果也返回多块4、使用SparkStreaming编写的程序与编写Spark程序非常相似5、在Spark程序中，主要通过操作RDD（ResilientDistributedDatasets弹性分布式数据集）提供的接口，如map、reduce、filter等，实现数据的批处理6、而在SparkStreaming中，则通过操作DStream（表示数据流的RDD序列）提供的接口，这些接口和RDD提供的接口类似SparkStreaming–编程模型DStream（DiscretizedStream）作为SparkStreaming的基础抽象，它代表持续性的数据流。这些数据流既可以通过外部输入源赖获取，也可以通过现有的Dstream的transformation操作来获得。在内部实现上，DStream由一组时间序列上连续的RDD来表示。每个RDD都包含了自己特定时间间隔内的数据流。DStream中在时间轴下生成离散的RDD序列：SparkStreaming–示例（worldcount）KafkaStreamingResultSparkStreaming–示例（访问用户数统计）实时统计当前小时开始当当前时刻为止，访问网站的用户数；有两个点注意：1、当前小时开始，实时统计在当前小时任意时刻的用户数2、用户数需要做去重处理查看代码：E_UserCountStreaming.scalaSpark性能优化原则一：避免创建重复的RDD原则二：尽可能复用同一个RDD原则三：对多次使用的RDD进行持久化原则四：尽量避免使用shuf