Spark_Java

1. Spark介绍
1.1. 什么是Spark

Apache Spark 是专为大规模数据处理而设计的快速通用的计算引擎。 Spark 是加州大学伯克利分校的 AMP实验室所开源的类 Hadoop MapReduce 的通用并行计算框架， Spark 拥有 Hadoop MapReduce 所具有的优点；但不同于 MapReduce 的是 Job 中间输出结果可以缓存在内存中，从而不再需要读写 HDFS ，减少磁盘数据交互，因此 Spark 能更好地适用于数据挖掘与机器学习等需要迭代的算法。 Spark 是 Scala 编写，方便快速编程。

1.2. Spark 与MR 的区别

都是分布式计算框架， Spark 计算中间结果基于内存缓存， MapReduce 基于 HDFS 存储。也正因此， Spark 处理数据的能力一般是 MR 的三到五倍以上， Spark 中除了基于内存计算这一个计算快的原因，还有 DAG（DAGShecdule）有向无环图来切分任务的执行先后顺序

2. SparkCore

2.1. Partition

2.1.1. 概念

Spark RDD 是一种分布式的数据集，由于数据量很大，因此要它被切分并存储在各个结点的分区当中
Spark中，RDD（Resilient Distributed Dataset）是其最基本的抽象数据集，其中每个RDD是由若干个Partition组成

2.1.2. 分区方式

Spark包含两种数据分区方式：HashPartitioner（哈希分区）和RangePartitioner（范围分区）。
HashPartitioner（哈希分区）
- 分区规则为：partitionId = Key.hashCode % numPartitions
- 有可能产生数据倾斜
RangePartitioner（范围分区）
- 解决数据倾斜问题
- 基于抽样的思想来对数据进行分区
- 蓄水池抽样法

2.1.3. HDFS与Partition

block位于存储空间、partition 位于计算空间
block的大小是固定的、partition 大小是不固定的
block是有冗余的、不会轻易丢失，partition（RDD）没有冗余设计、丢失之后重新计算得到

2.2. RDD

RDD(Resilient Distributed Dataset) 弹性分布式数据集。

partition 数量，大小没有限制，体现了 RDD 的弹性
RDD 是由 Partition 组成， partition 是分布在不同节点上的

2.2.1. RDD五大属性

一个RDD中有多个Partition组成
RDD中的每一个task运行在自己的Partition上
RDD需要依赖其他RDD
分区器必须作用于K V结构的RDD上
RDD会默认寻址最优的计算位置，一般计算向数据靠拢尽量少进行数据拉取的操作

2.2.2. Lineage血统

RDD 的最重要的特性之一就是血缘关系（Lineage )，它描述了一个 RDD 是如何从父 RDD 计算得来的。如果某个 RDD 丢失了，则可以根据血缘关系，从父 RDD 计算得来

2.3. 系统架构

Master ( standalone 模式)：资源管理的主节点（进程）。
Cluster Manager ：在集群上获取资源的外部服务（例如： standalone ； yarn ； mesos ）。
Worker ( standalone 模式）：资源管理的从节点（进程）或者说是是管理本机资源的进程。
Application ：基于 Spark 的用户程序，包含 driver 程序和运行在集群上的 executor 程序，即一个完整的 spark 应用。
Dirver （ program ）：用来连接工作进程（ worker ）的程序。
Executor ：是在一个 worker 进程所管理的节点上为某 Application 启动的一个个进程，这个进程负责运行任务，并且负责将数据存在内存或者磁盘上，每个应用之间都有各自独立的 executors 。
Task ：被发送到 executor 上的工作单元。
Job ：包含很多任务（ Task ）的并行计算，和 action 算子对应。
Stage ：一个 job 会被拆分成很多组任务，每组任务被称为 Stage （就像 MapReduce 分为 MapTask 和 ReduceTask 一样）。

3. 算子(单文件)

Spark 记录了 RDD 之间的生成和依赖关系。但是只有当 F 进行行动操作时，Spark 才会根据 RDD 的依赖关系生成 DAG，并从起点开始真正的计算。

3.1. 转换算子

概念
- Transformations 类算子叫做转换算子（本质就是函数）， Transformations 算子是延迟执行，也叫懒加载执行
常见 Transformation 类算子
- filter ：过滤符合条件的记录数， true 保留， false 过滤掉
- map ：将一个 RDD 中的每个数据项，通过 map 中的函数映射变为一个新的元素。特点：输入一条，输出一条数据
- flatMap ：先 map 后 flat 。与 map 类似，每个输入项可以映射为0到多个输出项
- sample 随机抽样算子，根据传进去的小数按比例进行有放回或者无放回的抽样。
- reduceByKey 将相同的 Key 根据相应的逻辑进行处理。
- sortByKey / sortBy 作用在 K,V格式的RDD 上，对 key 进行升序或者降序排序

3.3. 控制算子

概念
- 将 RDD 持久化，持久化的单位是 partition
- 控制算子有三种， cache , persist , checkpoint 。 cache 和 persist 都是懒执行的。必须有一个 action 类算子触发执行
- checkpoint 算子不仅能将 RDD 持久化到磁盘，还能切断 RDD 之间的依赖关系

3.3.1. cache

默认将 RDD 的数据持久化到内存中。 cache 是懒执行
cache() = persist() = persist(StorageLevel.Memory_Only)
rdd.cache().count() 返回的不是持久化的RDD，而是一个数值

object HelloSparkCatch {
    def main(args: Array[String]): Unit = {
        val sparkConf = new
SparkConf().setMaster("local").setAppName("SparkCatch" + System.currentTimeMillis())
        val sparkContext = new SparkContext(sparkConf)
        var lines =
sparkContext.textFile("src/main/resources/NASA_access_log_Aug95")
        lines = lines.cache();
        val startTime = System.currentTimeMillis
        val count = lines.count
        val endTime = System.currentTimeMillis
        System.out.println("第一次共" + count + "条数据，" + "计算时间=" + (endTime - startTime))
        val cacheStartTime = System.currentTimeMillis
        val cacheResult = lines.count
        val cacheEndTime = System.currentTimeMillis
        System.out.println("第二次共" + cacheResult + "条数据，" + "计算时间=" + (cacheEndTime - cacheStartTime))
        sparkContext.stop
    }
}

3.3.2. persist

可以指定持久化的级别。最常用的是 MEMORY_ONLY 和 MEMORY_AND_DISK
持久化级别如下：

3.3.3. checkpoint

checkpoint 将 RDD 持久化到磁盘，还可以切断 RDD 之间的依赖关系，也是懒执行
执行原理：
- 当 RDD 的 job 执行完毕后，会从 finalRDD 从后往前回溯。
- 当回溯到某一个 RDD 调用了 checkpoint 方法，会对当前的 RDD 做一个标记。
- Spark 框架会自动启动一个新的 job ，重新计算这个 RDD 的数据，将数据持久化到Checkpint 目录中。

使用 checkpoint 时常用优化手段：

对 RDD 执行 checkpoint 之前，最好对这个 RDD 先执行 cache
这样新启动的 job 只需要将内存中的数据拷贝到Checkpint目录中就可以，省去了重新计算这一步。

def main(args: Array[String]): Unit = {
    val sparkConf = new
SparkConf().setMaster("local").setAppName("SparkCheckPoint" +
System.currentTimeMillis())
    val sparkContext = new SparkContext(sparkConf)
    sparkContext.setCheckpointDir("./checkpoint")
    val lines: RDD[String] =
sparkContext.textFile("src/main/resources/NASA_access_log_Aug95")
    val words: RDD[String] = lines.flatMap(_.split(" "))
    println("words" + words.getNumPartitions)
    words.checkpoint
    words.count
    sparkContext.stop
}

Spark

Java相关栏目本月热门文章