SparkCore杂记一

r囧r小猫 2022-06-06 11:22 190阅读 0赞

接触Spark也有一段时间了，最开始一直都是上网看一些博客，自己搭建个虚拟机倒腾，都是一些零散的学习，回头想想还是有必要系统性的学习、理解一遍，本系列博客将会参照spark 官方文档上来一一讲解，但是也不会仅仅只是翻译（翻译也不全面），其中还会加上一些自己的理解、在项目中的一些所见所闻，希望自己能够慢慢成长起来，如果理解有误还请各位指正。

Spark是基于内存计算的分布式处理系统，相对于Hadoop中的mapreduce来说，个人理解三点：一是运行速度快、二是容错处理、三是编程方便，这个之后会慢慢讲解，Spark最主要的抽象概念就是RDD(resilient distributed dataset ,弹性分布式数据集)，具有以下几个特点：

1. 分区 ，表示RDD的数据是存在于多个节点上的，可以并行计算

2. 可基于内存的，之前听闻很多同学直观理解RDD就是内存数据，其实这个理解是不太准确的， RDD是亦可基于内存也可基于磁盘，默认的持久化机制是存储在内存中，但是在实际是我们往往不会这样使用，如果分配的内存写满了，那么部分数据就会被清除掉，所以通常选择同时内存与磁盘进行持久化。RDD持久化的级别是有多种的在后面将会一一讲解。

3. 容错的 ，容错简单的来说当RDD执行到某一步出错，我们还能恢复到当前的数据上来，讲到容错就不得不说RDD之间的依赖，我们都知道一个RDD（子RDD）可以由另外一个RDD(父RDD)转换过来，就存在前后RDD之间的依赖，分为：宽依赖与窄依赖，宽依赖表示父RDD与子RDD是一对多的关系，通常由reduceByKey这样的算子操作产生（会使key进行重新分配，数据需要跨节点传输），窄依赖表是父RDD与子RDD是一对一的关系，通常有map这样的算子操作产生（key不会重新分配，仅仅是在当前节点操作），当窄依赖的RDD出错，只需要重新计算当前分区的数据即可，当宽依赖RDD出错，在重新计算当前分区的数据的同时，还会重新计算其他分区的数据，这样就会产胜冗余数据，计算的性能就会下降。对于数据的恢复常见的有两种方式：记录检查点，在某时刻把当前的数据保存下来，以后的计算出错了，就直接来获取；数据更新操作记录，记录对数据的每一步操作，出错了就重头来一遍即可。RDD中本身有血链（lineage）,会记录对RDD上所有的操作，那么出错是重新计算即可，但是对于宽依赖的这种RDD，重新计算代价比较大（数据冗余、跨节点数据传输），提供了记录检查点方式，即checkpoint ,Spark中RDD checkpoint 表示会从RDD的产生到当前的RDD执行一遍生成的数据存放在由sparkContext.setCheckPointDir()的目录中，那么当之后的任务执行失败直接获取当前的数据即可，不需要重新计算，由于checkpoint 相当于由执行了一遍RDD 算子操作，那么在使用checkpoint之前，通常会将RDD持久化一遍，那么直接从内存或者磁盘中获取数据将数据保存到指定的目录中(一般是hdfs上)，同时会将当前的RDD设置成为父RDD,清楚它更高层次的依赖，在这里你可能会想到RDD进行持久化不能进行容错么，不能，首先一点RDD持久化是对多次访问一个RDD上的性能优化，另外它不会成为父RDD并且不会清楚更高层次的依赖，当出错还是会沿着lineage往上寻找。

4. 只读的，直观理解RDD是不能进行update的

Spark还有另外一个抽象概念，那就是共享变量，我们都明白Spark 任务是在多个Executor中（多个不同的进程中）运行的(我们这里指的是非local模式)，Driver(理解为客户端)将代码通过序列化方式发送到各个executor中执行，如果我们要实现不同Executor之间数据共享那么就要使用共享变量，在讲解变量共享方式之前需要先来理解闭包原则，闭包原则指的是每个Executor中的数据对外是不可见的，除非你使用rdd.collect() 将rdd的数据发送到了driver端，闭包具体在编程上体现：foreache、map等操作上，foreach、map里面使用的变量对外都是不可见的。Spark解决共享变量有两种方式：`Accumulator (原子变量)、 broadcast(广播变量) ，原子变量表示在executor中对变量的操作对于另外`一个executo是可见的，使用方式sparkContext.longAccumulator()得到一个原子变量，然后在map、reduce等操作中使用，官方文档中给出了计数的例子，非常形象，原子变量不难理解，可类似于java中的AtomicLong的使用，我们都知道如果对一个long 类型的数据（初始值为0）使用多线程做1000次++，那么我们得到的结果很有可能是小于1000的，使用AtomicLong类型确保了其原子性。Saprk中自带的ongAccumulator中add方式仅仅是做加法， 我们也可以自定义原子变量，只需要实现AccumulatorParam<T>接口即可，重写接口方法，实现自己的业务逻辑；广播变量表示将数据以序列化的方式发送到每个Executor中,这类数据通常比较固定，不会修改的，那么在每个Executor中就会存在一份相同的数据，使用sparkContext.broadcast(v)得到广播变量，在map/reduce中直接使用即可。

**    Spark的运行模式包括local、standalone、yarn、mesos,关于其如何部署、之间的差异可以参考我**

`转`**`载的其他博文或者参考官方文档更好，我在实际生产中使用的是yarn模式的。`RDD的产生通过**

**sparkContext.parallelize(data)、sparkContext.textFile(path)这两种方式获得，第一种data表示**一个现有的数据集，第二种path表示hdfs、localpath(每个节点上都要存放一份)或者AS3路径上读取，在实际中通常指hdfs上的一个路径（可以是路劲、文件、\*.txt、压缩.gz格式）。parallelize与textFile还有第二个可选的参数:分区数，指定初始RDD的分区数量，在默认情况下对于parallelize，如果我们指定了

**spark.default.parallelism(默认情况下一个stage的并行任务个数，后面使用defaultParallelism指**

**代)，则使用该值，如果没有，则去max(count(core个数)，2)(yarn情况)；对于textFile,默认取**

**min(defaultParallelism,2),对于hdfs上数据，则其分区的数量是文件分片的数量，如果要重新指定值则其值**不能小于当前文件的分片数量。对于RDD分区我们尽量遵守以下两个原则：1. 分区的个数等于core的总数，资源的最大化利用；2. 分区中的数量尽量保持均匀，对于每个任务来说执行的时间就比较均匀，最主要的是不会产生数据倾斜。 RDD提供了两个方法进行重新分区，rdd.repartition(num),

**rdd.coalesce(num,isShuffle),repartition是coalesce的isShuffle为true的用法，即执行shuffle,**当我们需要增大RDD的分区个数的时候，就必须执行shuffle,需要减少RDD的分区个数就将isShuffle设置成为false,避免shuffle，提高效率，通常再有很多个小的任务时候，我们可以减少RDD的分区个数。Spark中提供了连个分区类HashPartitioner/RangerPartitioner,HashPartitioner将RDD中每条记录中的key进行hash,**然后除以分区数量取余，得到分区编号，它可能会导致数据分布不均匀，RangeParitioner范围分区，一定key范****围的记录在同一分区中，在实际中一般使用默认的HashPartitioner，当然也可以自定义Partitioner.关于二者具体可参考https://www.iteblog.com/archives/1522.html。在实际的使用中，还会读取hbase中的数据生成RDD,使用sparkContext.newAPIHadoopRDD(),得到****JavaPairRDD<ImmutableBytesWritable, Result>类型的RDD。**

**   通过以上方式得到RDD可以对其使用两种类型的操作，一种是transformations(转换)，由一种RDD经过变换得到另外一种RDD,例如map/filter/reduceByKey/repartition;另外一种是actions（动作），各个Executor将结果汇总输出到Driver段，例如collect/foreache/reduce,transformations是Lazy(懒惰得)，不会立即执行，只有当遇到actions时，才会触发RDD操作.我们在使用的过程中可能会发现在reduceByKey里面执行System.out输出操作时，但是在我们的客户端却没有输出，那是什么reduceByKey这些操作是在executor中执行，这个输出会打印到相应executor的日志文件中，而我们一般看的是Driver端，要想在Driver端看到打印信息可以使用rdd.foreache中打印、rdd.collect().take()输出。关于transformations与actions具体包括哪些可直接参考官方文档，这里不做一一讲解。**