点一下关注吧！！！非常感谢！！持续更新！！！

目前已经更新到了：

Hadoop（已更完）HDFS（已更完）MapReduce（已更完）Hive（已更完）Flume（已更完）Sqoop（已更完）Zookeeper（已更完）HBase（已更完）Redis （已更完）Kafka（已更完）Spark（正在更新！）

章节内容

上节完成的内容如下：

Spark RDD的依赖关系重回 WordCountRDD 持久化RDD 缓存

RDD容错机制

基本概念

涉及到的算子：checkpoint，也是Transformation

Spark中对于数据的保存除了持久化操作外，还提供了检查点的机制检查点本质是通过RDD写入高可靠的磁盘，主要目的是为了容错。检查点通过将数据写入到HDFS文件系统实现了RDD的检查点功能。Lineage过长会造成容错成本过高，这样就不如在中间阶段做检查点容错，如果之后有节点出现问题而丢失分区，从做检查点的RDD开始重做Lineage，就会减少开销cache和checkpoint是有显著区别的，缓存把RDD计算出来然后放到内存中，但RDD的依赖链不能丢掉，当某个点某个Executor宕机了，上面cache的RDD就会丢掉，需要通过依赖链重新计算。不同的是，checkpoint是把RDD保存在HDFS中，是多副本的可靠存储，此时依赖链可以丢弃，所以斩断了依赖链。

适合场景

DAG中的Lineage过长，如果重新计算，开销会很大在宽依赖上做checkpoint获得的收益更大

启动Shell

<code># 启动 spark-shell

spark-shell --master local[*]

checkpoint

// 设置检查点目录

sc.setCheckpointDir("/tmp/checkpoint")

val rdd1 = sc.parallelize(1 to 1000)

val rdd2 = rdd1.map(_*2)

rdd2.checkpoint

// checkpoint是lazy操作

rdd2.isCheckpointed

可以发现，返回结果是False

RDD 依赖关系1

checkpoint之前的rdd依赖关系

rdd2.dependencies(0).rddrdd2.dependencies(0).rdd.collect

我们可以观察到，依赖关系是有的，关系到之前的 rdd1 的数据了：

触发checkpoint

我们可以通过执行 Action 的方式，来触发 checkpoint

执行一次action，触发checkpoint的执行

rdd2.countrdd2.isCheckpointed

此时观察，可以发现 checkpoint 已经是 True 了：

RDD依赖关系2

我们再次观察RDD的依赖关系：

再次查看RDD的依赖关系。可以看到checkpoint后，RDD的lineage被截断，变成从checkpointRDD开始

rdd2.dependencies(0).rddrdd2.dependencies(0).rdd.collect

此时观察到，已经不是最开始的 rdd1 了：

查看checkpoint

我们可以查看对应的保存的文件，查看RDD所依赖的checkpoint文件

rdd2.getCheckpointFile

运行的结果如下图：

RDD的分区

基本概念

<code>spark.default.paralleism: 默认的并发数 2

本地模式

# 此时 spark.default.paralleism 为 N

spark-shell --master local[N]

# 此时 spark.default.paralleism 为 1

spark-shell --master local

伪分布式

x为本机上启动的Executor数y为每个Executor使用的core数z为每个Executor使用的内存spark.default.paralleism 为 x * y

spark-shell --master local-cluster[x,y,z]

分布式模式

spark.default.paralleism = max(应用程序持有Executor的core总数, 2)

创建RDD方式

集合创建

简单的说，RDD分区数等于cores总数

val rdd1 = sc.paralleize(1 to 100)

rdd.getNumPartitions

textFile创建

如果没有指定分区数：

本地文件： rdd的分区数 = max(本地文件分片数，sc.defaultMinPartitions)HDFS文件：rdd的分区数 = max(HDFS文件block数，sc.defaultMinPartitions)

需要额外注意的是：

本地文件分片数 = 本地文件大小 / 32M读取 HDFS 文件，同时指定了分区数 < HDFS文件的Block数，指定的数将不会生效

val rdd = sc.textFile("data/1.txt")

rdd.getNumPartitions

RDD分区器

判断分区器

以下RDD分别是否有分区器，是什么类型的分区器

val rdd1 = sc.textFile("/wcinput/wc.txt")

rdd1.partitioner

val rdd2 = sc.flatMap(_.split("\\s+"))

rdd2.partitioner

val rdd3 = rdd2.map((_, 1))

rdd3.partitioner

val rdd4 = rdd3.reduceByKey(_ + _)

rdd4.partitioner

val rdd5 = rdd4.sortByKey()

rdd5.partitioner

分区器作用与分类

在PairRDD(key,value)中，很多操作都是基于Key的，系统会按照Key对数据进行重组，如 GroupByKey

数据重组需要规则，最常见的就是基于Hash的分区，此外还有一种复杂的基于抽样Range分区方法：

HashPartitioner

最简单、最常用，也是默认提供的分区器。

对于给定的Key，计算HashCode，并除以分区的个数取余，如果余数小于0，则用余数+分区的个数，最后返回的值就是这个Key所属的分区ID。

该分区方法可以保证Key相同的数据出现在同一个分区中。

用户可以通过 partitionBy主动使用分区器，通过 partitions参数指定想要分区的数量。

默认情况下的分区情况是：

<code>val rdd1 = sc.makeRDD(1 to 100).map((_, 1))

rdd1.getNumPartitions

执行结果如下图所示：

执行结果如下图所示，分区已经让我们手动控制成10个了：

<code>val rdd2 = rdd1.partitionBy(new org.apache.spark.HashPartitioner(10))

rdd2.getNumPartitions

rdd2.glom.collect.foreach(x => println(x.toBuffer))

RangePartitioner

简单来说就是将一定范围内的数映射到某个分区内，在实现中，分界的算法尤为重要，用到了水塘抽样算法。sortByKey会使用RangePartitioner。

进行代码的测试：

<code>val rdd3 = rdd1.partitionBy(new org.apache.spark.RangePartitioner(10, rdd1))

rdd3.glom.collect.foreach(x => println(x.toBuffer))

执行结果如下图所示：

但是现在的问题是：在执行分区之前其实并不知道数据的分布情况，如果想知道数据的分区就需要对数据进行采样。

Spark中的RangePartitioner在对数据采样的过程中使用了 “水塘采样法”水塘采样法是：在包含N个项目的集合S中选取K个样本，其中N为1或者很大的未知的数量，尤其适用于不能把所有N个项目都存放到主内存的情况。在采样过程中执行了 collect() 操作，引发了 Action 操作。

自定义分区器

Spark允许用户通过自定义的Partitioner对象，灵活的来控制RDD的分区方式。

我们需要实现自定义分区器，按照以下的规则进行分区：

分区 0 < 100100 <= 分区1 < 200200 <= 分区2 < 300300 <= 分区3 < 400…900 <= 分区9 < 1000

编写代码

<code>package icu.wzk

import org.apache.spark.rdd.RDD

import org.apache.spark.{ Partitioner, SparkConf, SparkContext}

import scala.collection.immutable

class MyPartitioner(n: Int) extends Partitioner {

override def numPartitions: Int = n

override def getPartition(key: Any): Int = {

val k = key.toString.toInt

k / 100

}

}

object UserDefinedPartitioner {

def main(args: Array[String]): Unit = {

val conf = new SparkConf()

.setAppName("UserDefinedPartitioner")

.setMaster("local[*]")

val sc = new SparkContext(conf)

sc.setLogLevel("WARN")

val random = scala.util.Random

val arr: immutable.IndexedSeq[Int] = (1 to 100)

.map(idx => random.nextInt(1000))

val rdd1: RDD[(Int, Int)] = sc.makeRDD(arr).map((_, 1))

rdd1.glom.collect.foreach(x => println(x.toBuffer))

println("=========================================")

val rdd2 = rdd1.partitionBy(new MyPartitioner(10))

rdd2.glom.collect().foreach(x => println(x.toBuffer))

sc.stop()

}

}

打包上传

这里之前已经重复过多次，就跳过了

mvn clean package

运行测试

spark-submit --master local[*] --class icu.wzk.UserDefinedPartitioner spark-wordcount-1.0-SNAPSHOT.jar

可以看到如下的运行结果：