Spark计算引擎

一、Spark简介

• 由加州大学伯克利分校的AMP实验室开源
• 大规模分布式通用计算引擎
• 具有高吞吐、低延时、通用易扩展、高容错等特点
• 使用Scala语言开发，提供了丰富的开发API，支持Scala、Java、
• Python、R开发语言
• Spark提供多种运行模式
  Spark架构

二、Spark特点

• 计算高效
  • 使用内存计算引擎，提供Cache缓存机制支持迭代计算或多次数据共享，减少数据读取的IO开销
  • DAG引擎，减少多次计算之间中间结果写到HDFS的开销
  • 使用多线程池模型来减少task启动开销，shuffle过程中避免不必要的
  • sort操作以及减少磁盘IO操作
• 通用易用
  • 提供了丰富的开发API，支持Scala、Java、Python、R开发语言
  • 集成批处理、流处理、交互式计算、机器学习算法、图计算
• 运行模式多样
  • Local、Standalone、Yarn、Mesos
    运行比对图

三、Spark核心概念-RDD

• RDD：Resilient Distributed Datasets弹性分布式数据集
  • Spark基于RDD进行计算
  • 分布在集群中的只读对象集合（由多个Partition构成）
  • 可以存储在磁盘或内存中
  • 可以通过并行转换操作构造
  • 失效后自动重构
    RDD

1.RDD操作

• Transformation
  • 将Scala集合或者Hadoop数据集构造一个新的RDD
  • 通过已有的RDD产生新的RDD
  • 只记录转换关系，不触发计算
  • 如：map、filter等
    -Action
  • 通过RDD计算得到一个或者一组值
  • 真正触发执行
  • 如：count、collect、saveAsTextFile

RDD操作

2.RDD操作示例

rdd1.map(_+1).saveAsTextFile("hdfs://node01:9000/")

RDD输出

3.Transformation与Action对比

• 接口定义方式不同
  • Transformation：RDD[X] -> RDD[Y]
  • Action：RDD[X] -> Z
• 执行计算方式不同
  • Transformation采用惰性执行方式，只记录RDD转化关系，不会触发真正计算执行
  • Action真正触发计算执行

4.Transformation Lazy Execution

• 惰性执行

懒惰执行

5.程序执行流程

• Spark中的WordCount

val rdd1 = sc.textFile("hdfs://192.168.183.101:9000/data/wc/in")
val rdd2 = rdd1.flatMap(_.split("\t"))
val rdd3= rdd2.map((_,1))
val rdd4 = rdd3.reduceByKey(_ + _)
rdd4.saveAsTextFile(“hdfs://192.168.183.100:9000/data/wc/out”)

Spark程序执行流程

6.RDD Dependency依赖

• Narrow Dependency窄依赖
  • 父RDD中的分区最多只能被一个子RDD的一个分区使用
  • 子RDD如果只有部分分区数据丢失 或者损坏只需要从对应的父RDD重新计算恢复
• Shuffle Dependency宽依赖
  • 子RDD分区依赖父RDD所有分区
  • 子RDD如果部分分区或者全部分区 数据丢失或者损坏需要从所有父RDD重新计算，相对窄依赖付出的
    代价更高，尽量避免宽依赖的使用

窄依赖
宽依赖

7.RDD Cache缓存

• Spark允许将RDD缓存到内存或磁盘上，方便重用，提高性能
• Spark提供了多种缓存级别
• 用户可以根据实际需求进行调整

val rdd = sc.textFile(inputArg)
rdd.cache()//实际上是调用了rdd.persist(StorageLevel.MEMORY_ONLY)
//data.persist(StorageLevel.MEMORY_AND_DISK)

RDD缓存

四、Spark程序架构

• Driver：一个Spark程序有一个Driver，一个Driver创建一个Spark Context，程序的main
  函数运行在Driver中。Driver主要负责Spark程序的解析、划分Stage，调度Task到Executor上执行
• SparkContext：加载配置信息，初始化Spark程序运行环境，创建内部的DAGScheduler和TaskScheduler
• Executor：负责执行Driver分发的Task任务，集群中一个节点可以启动多个Executor，每个一个Executor通过多线程运行多个Task任务
• Task：Spark运行的基本单位，一个Task负责处理RDD一个分区的计算逻辑
  Spark程序架构

五、Spark运行模式

• Local本地模式
  • 单机运行，通常用于测试
• Standalone独立模式
  • Spark集群单独运行
• Yarn/Mesos
  • 运行在其他资源管理系统上，如Yarn、Mesos

1.Spark Local模式

• Local本地模式
  • 将Spark应用以多线程方式，直接运行在本地，方便调试
• 本地模式分类

local：只启动一个线程运行executor
local[n]：启动n个线程运行executor
local[*]：启动跟cpu数目相同的executor

2.Spark Standalone模式

• 集群独立部署
  Spark standalone模式

3.YARN程序运行流程

Yarn执行流程

4.Spark YARN模式

• yarn-cluster

Spark on yarn

5.Spark内部执行流程

Spark 内部执行流程

• 生成逻辑查询计划

sc.textFile(inputArg)
.flatMap(_.split("\t"))
.map((_,1))
.reduceByKey(_ + _)
.saveAsTextFile(outArg)

逻辑查询计划

• 生成物理查询计划
  物理查询计划
• 任务调度与执行
  任务调度与执行

六、Spark调度模块

• DAG：Directed Acyclic Graph有向无环图
• DAGScheduler
  • 根据计算任务的依赖关系建立DAG
  • 根据依赖关系是否是宽依赖，将DAG划分为不同的Stage阶段
  • 将各个Stage中的Task组成的TaskSet提交到TaskScheduler
• TaskScheduler
  • 负责Application的Job调度
  • 重新提交执行失败的Task
  • 为执行速度慢的Task启动备份任务
    Sparkr 模块调度

1.Spark任务类型和Job划分

• Spark中task类型
  • ShuffleMapTask：除了最后一个输出Task，其他Task类型都是 ShuffleMapTask
  • ResultTask：只有最后一个阶段输出的Task是ResultTask
• Appication中调用一次Action就会生成一个Job
  Spark 任务划分