Spark修炼之道(进阶篇)——Spark入门到精通:第四节 Spark编程模型(一)

  1. 云栖社区>
  2. 博客>
  3. 正文

Spark修炼之道(进阶篇)——Spark入门到精通:第四节 Spark编程模型(一)

周志湖 2015-09-20 19:43:00 浏览3603
展开阅读全文

作者:周志湖
网名:摇摆少年梦
微信号:zhouzhihubeyond

本节主要内容

  1. Spark重要概念
  2. 弹性分布式数据集(RDD)基础

1. Spark重要概念

本节部分内容源自官方文档:http://spark.apache.org/docs/latest/cluster-overview.html

(1)Spark运行模式

目前最为常用的Spark运行模式有:
- local:本地线程方式运行,主要用于开发调试Spark应用程序
- Standalone:利用Spark自带的资源管理与调度器运行Spark集群,采用Master/Slave结构,为解决单点故障,可以采用ZooKeeper实现高可靠(High Availability,HA)
- Apache Mesos :运行在著名的Mesos资源管理框架基础之上,该集群运行模式将资源管理交给Mesos,Spark只负责进行任务调度和计算
- Hadoop YARN : 集群运行在Yarn资源管理器上,资源管理交给Yarn,Spark只负责进行任务调度和计算
Spark运行模式中Hadoop YARN的集群运行方式最为常用,本课程中的第一节便是采用Hadoop YARN的方式进行Spark集群搭建。如此Spark便与Hadoop生态圈完美搭配,组成强大的集群,可谓无所不能。

(2)Spark组件(Components)

一个完整的Spark应用程序,如前一节当中SparkWordCount程序,在提交集群运行时,它涉及到如下图所示的组件:
这里写图片描述

各Spark应用程序以相互独立的进程集合运行于集群之上,由SparkContext对象进行协调,SparkContext对象可以视为Spark应用程序的入口,被称为driver program,SparkContext可以与不同种类的集群资源管理器(Cluster Manager),例如Hadoop Yarn、Mesos等 进行通信,从而分配到程序运行所需的资源,获取到集群运行所需的资源后,SparkContext将得到集群中其它工作节点(Worker Node) 上对应的Executors (不同的Spark应用程序有不同的Executor,它们之间也是独立的进程,Executor为应用程序提供分布式计算及数据存储功能),之后SparkContext将应用程序代码分发到各Executors,最后将任务(Task)分配给executors执行。

Term(术语) Meaning(解释)
Application(Spark应用程序) 运行于Spark上的用户程序,由集群上的一个driver program(包含SparkContext对象)和多个executor线程组成
Application jar(Spark应用程序JAR包) Jar包中包含了用户Spark应用程序,如果Jar包要提交到集群中运行,不需要将其它的Spark依赖包打包进行,在运行时
Driver program 包含main方法的程序,负责创建SparkContext对象
Cluster manager 集群资源管理器,例如Mesos,Hadoop Yarn
Deploy mode 部署模式,用于区别driver program的运行方式:集群模式(cluter mode),driver在集群内部启动;客户端模式(client mode),driver进程从集群外部启动
Worker node 工作节点, 集群中可以运行Spark应用程序的节点
Executor Worker node上的进程,该进程用于执行具体的Spark应用程序任务,负责任务间的数据维护(数据在内存中或磁盘上)。不同的Spark应用程序有不同的Executor
Task 运行于Executor中的任务单元,Spark应用程序最终被划分为经过优化后的多个任务的集合(在下一节中将详细阐述)
Job 由多个任务构建的并行计算任务,具体为Spark中的action操作,如collect,save等)
Stage 每个job将被拆分为更小的task集合,这些任务集合被称为stage,各stage相互独立(类似于MapReduce中的map stage和reduce stage),由于它由多个task集合构成,因此也称为TaskSet

2. 弹性分布式数据集(RDD)基础

弹性分布式数据集(RDD,Resilient Distributed Datasets),由Berkeley实验室于2011年提出,原始论文名字:Resilient Distributed Datasets: A Fault-Tolerant Abstraction for In-Memory Cluster Computing 原始论文非常值得一读,是研究RDD的一手资料,本节内容大部分将基于该论文。

(1)RDD设计目标

RDD用于支持在并行计算时能够高效地利用中间结果,支持更简单的编程模型,同时也具有像MapReduce等并行计算框架的高容错性、能够高效地进行调度及可扩展性。RDD的容错通过记录RDD转换操作的lineage关系来进行,lineage记录了RDD的家族关系,当出现错误的时候,直接通过lineage进行恢复。RDD最合数据挖掘, 机器学习及图计算,因此这些应用涉及到大家的迭代计算,基于内存能够极大地提升其在分布式环境下的执行效率;RDD不适用于诸如分布式爬虫等需要频繁更新共享状态的任务。

下面给出的是在spark-shell中如何查看RDD的Lineage

//textFile读取hdfs根目录下的README.md文件,然后筛选出所有包括Spark的行
scala> val rdd2=sc.textFile("/README.md").filter(line => line.contains("Spark"))
rdd2: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[2] at filter at <console>:21
//toDebugString方法会打印出RDD的家族关系
//可以看到textFile方法会生成两个RDD,分别是HadoopRDD
//MapPartitionsRDD,而filter同时也会生成新的MapPartitionsRDD
scala> rdd2.toDebugString
15/09/20 01:35:27 INFO mapred.FileInputFormat: Total input paths to process : 1
res0: String = 
(2) MapPartitionsRDD[2] at filter at <console>:21 []
 |  MapPartitionsRDD[1] at textFile at <console>:21 []
 |  /README.md HadoopRDD[0] at textFile at <console>:21 []

(2)RDD抽象

RDD在Spark中是一个只读的(val类型)、经过分区的记录集合。RDD在Spark中只有两种创建方式:(1)从存储系统中创建;(2)从其它RDD中创建。从存储中创建有多种方式,可以是本地文件系统,也可以是分布式文件系统,还可以是内存中的数据。
下面的代码演示的是从HDFS中创建RDD

scala> sc.textFile("/README.md")
res1: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[4] at textFile at <console>:22

下面的代码演示的是从内存中创建RDD

//内存中定义了一个数组
scala> val data = Array(1, 2, 3, 4, 5)
data: Array[Int] = Array(1, 2, 3, 4, 5)
//通过parallelize方法创建ParallelCollectionRDD
scala> val distData = sc.parallelize(data)
distData: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[5] at parallelize at <console>:23

下面的代码演示的是从其它RDD创建新的RDD

//filter函数将distData RDD转换成新的RDD
scala> val distDataFiletered=distData.filter(e=>e>2)
distDataFiletered: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[6] at filter at <console>:25
//触发action操作(后面我们会讲),查看过滤后的内容
//注意collect只适合数据量较少时使用
scala> distDataFiltered.collect
res3: Array[Int] = Array(3, 4, 5)

(3)RDD编程模型

在前面的例子中,我们已经接触过到如何利用RDD进行编程,前面我们提到的

//filter函数将distData RDD转换成新的RDD
scala> val distDataFiletered=distData.filter(e=>e>2)
//触发action操作(后面我们会讲),查看过滤后的内容
//注意collect只适合数据量较少时使用
scala> distDataFiltered.collect

这段代码它已经给我们解释了RDD编程模型的核心思想:“filter函数将distData RDD转换成新的RDD”,“触发action操作”。也就是说RDD的操作包括Transformations(转换)、Actions两种。

transformations操作会将一个RDD转换成一个新的RDD,需要特别注意的是所有的transformation都是lazy的,如果对scala中的lazy了解的人都知道,transformation之后它不会立马执行,而只是会记住对相应数据集的transformation,而到真正被使用的时候才会执行,例如distData.filter(e=>e>2) transformation后,它不会立即执行,而是等到distDataFiltered.collect方法执行时才被执行,如下图所示
这里写图片描述
从上图可以看到,在distDataFiltered.collect方法执行之后,才会触发最终的transformation执行。

从transformation的介绍中我们知道,action是解决程序最终执行的诱因,action操作会返回程序执行结果如collect操作或将运行结果保存,例如SparkWordCount中的saveAsTextFile方法。

Spark 1.5.0支持的transformation包括:

(1)map
map函数方法参数:

/**
   * Return a new RDD by applying a function to all elements of this RDD.
   */
  def map[U: ClassTag](f: T => U): RDD[U]

//使用示例

scala> val rdd1=sc.parallelize(Array(1,2,3,4)).map(x=>2*x).collect
rdd1: Array[Int] = Array(2, 4, 6, 8)

(2)filter
方法参数:

/**
   * Return a new RDD containing only the elements that satisfy a predicate.
   */
  def filter(f: T => Boolean): RDD[T]

使用示例

scala> val rdd1=sc.parallelize(Array(1,2,3,4)).filter(x=>x>1).collect
rdd1: Array[Int] = Array(2, 3, 4)

(3)flatMap
方法参数:

/**
   *  Return a new RDD by first applying a function to all elements of this
   *  RDD, and then flattening the results.
   */
  def flatMap[U: ClassTag](f: T => TraversableOnce[U]): RDD[U] 

使用示例:

scala>  val data =Array(Array(1, 2, 3, 4, 5),Array(4,5,6))
data: Array[Array[Int]] = Array(Array(1, 2, 3, 4, 5), Array(4, 5, 6))

scala> val rdd1=sc.parallelize(data)
rdd1: org.apache.spark.rdd.RDD[Array[Int]] = ParallelCollectionRDD[2] at parallelize at <console>:23

scala> val rdd2=rdd1.flatMap(x=>x.map(y=>y))
rdd2: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[3] at flatMap at <console>:25

scala> rdd2.collect
res0: Array[Int] = Array(1, 2, 3, 4, 5, 4, 5, 6)


(4)mapPartitions(func)
本mapPartitions例子来源于:https://www.zybuluo.com/jewes/note/35032
mapPartitions是map的一个变种。map的输入函数是应用于RDD中每个元素,而mapPartitions的输入函数是应用于每个分区,也就是把每个分区中的内容作为整体来处理的。它的函数定义为:

def mapPartitions[U: ClassTag](f: Iterator[T] => Iterator[U], preservesPartitioning: Boolean = false): RDD[U]

f即为输入函数,它处理每个分区里面的内容。每个分区中的内容将以Iterator[T]传递给输入函数f,f的输出结果是Iterator[U]。最终的RDD由所有分区经过输入函数处理后的结果合并起来的。

scala> val a = sc.parallelize(1 to 9, 3)
scala> def myfunc[T](iter: Iterator[T]) : Iterator[(T, T)] = {
    var res = List[(T, T)]() 
    var pre = iter.next 
    while (iter.hasNext) {
        val cur = iter.next; 
        res .::= (pre, cur)
        pre = cur;
    } 
    res.iterator
}
scala> a.mapPartitions(myfunc).collect
res0: Array[(Int, Int)] = Array((2,3), (1,2), (5,6), (4,5), (8,9), (7,8))

上述例子中的函数myfunc是把分区中一个元素和它的下一个元素组成一个Tuple。因为分区中最后一个元素没有下一个元素了,所以(3,4)和(6,7)不在结果中。
mapPartitions还有些变种,比如mapPartitionsWithContext,它能把处理过程中的一些状态信息传递给用户指定的输入函数。还有mapPartitionsWithIndex,它能把分区的index传递给用户指定的输入函数。

(5)mapPartitionsWithIndex

mapPartitionsWithIndex函数是mapPartitions函数的一个变种,它的函数参数如下:

def mapPartitionsWithIndex[U: ClassTag](
f: (Int, Iterator[T]) => Iterator[U],
preservesPartitioning: Boolean = false): RDD[U]

```
scala> val a = sc.parallelize(1 to 9, 3)
//函数带分区索引,返回的集合第一个元素为分区索引
scala> def myfunc[T](index:T,iter: Iterator[T]) : Iterator[(T,T,T)] = {
    var res = List[(T,T, T)]() 
    var pre = iter.next 
    while (iter.hasNext) {
        val cur = iter.next
        res .::= (index,pre, cur) 
        pre = cur
    } 
    res.iterator
}
scala> a.mapPartitionsWithIndex(myfunc).collect
res11: Array[(Int, Int, Int)] = Array((0,2,3), (0,1,2), (1,5,6), (1,4,5), (2,8,9), (2,7,8))

这里写图片描述

(6)sample
方法参数:

 /**
   * Return a sampled subset of this RDD.
   *
   * @param withReplacement can elements be sampled multiple times (replaced when sampled out)
   * @param fraction expected size of the sample as a fraction of this RDD's size
   *  without replacement: probability that each element is chosen; fraction must be [0, 1]
   *  with replacement: expected number of times each element is chosen; fraction must be >= 0
   * @param seed seed for the random number generator
   */
  def sample(
      withReplacement: Boolean,
      fraction: Double,
      seed: Long = Utils.random.nextLong): RDD[T] 

使用示例:

scala> val a = sc.parallelize(1 to 9, 3)
a: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[12] at parallelize at <console>:21

scala> val smapledA=a.sample(true,0.5)
smapledA: org.apache.spark.rdd.RDD[Int] = PartitionwiseSampledRDD[13] at sample at <console>:23
scala> smapledA.collect
res12: Array[Int] = Array(3, 3, 3, 5, 6, 8, 8)

scala> val smapledA2=a.sample(false,0.5).collect
smapledA2: Array[Int] = Array(1, 4)

这里写图片描述

添加公众微信号,可以了解更多最新Spark、Scala相关技术资讯
这里写图片描述

网友评论

登录后评论
0/500
评论
周志湖
+ 关注