Spark学习笔记（一）——基础和架构

Spark基础和架构

一、Spark和Hadoop的比较

• MapReduce编程模型的局限性
  • 繁杂
    • 只有Map和Reduce两个操作，复杂的逻辑需要大量的样板代码
  • 处理效率低：
    • Map中间结果写磁盘，Reduce写HDFS，多个Map通过HDFS交换数据
    • 任务调度与启动开销大
  • 不适合迭代处理、交互式处理和流式处理
• Spark是类Hadoop MapReduce的通用并行框架
  • Job中间输出结果可以保存在内存，不再需要读写HDFS
  • 比MapReduce平均快10倍以上

二、Spark优势

• 速度快
  • 基于内存数据处理，比MR快100个数量级以上（逻辑回归算法测试）
  • 基于硬盘数据处理，比MR快10个数量级以上
• 易用性
  • 支持Java、Scala、Python、R语言
  • 交互式shell方便开发测试
• 通用性
  • 一栈式解决方案：批处理、交互式查询、实时流处理、图计算及机器学习
• 多种运行模式
  • YARN、Mesos、EC2、Kubernetes、Standalone、Local

三、Spark技术栈

• Spark Core

  • 核心组件，分布式计算引擎

  包含Spark的基本功能；尤其是定义RDD的API、操作以及这两者上的动作。其他Spark的库都是构建在RDD和Spark Core之上的

• Spark SQL

  • 高性能的基于Hadoop的SQL解决方案

  提供通过Apache Hive的SQL变体Hive查询语言（HiveQL）与Spark进行交互的API。每个数据库表被当做一个RDD，Spark SQL查询被转换为Spark操作

• Spark Streaming

  • 可以实现高吞吐量、具备容错机制的准实时流处理系统

  对实时数据流进行处理和控制。Spark Streaming允许程序能够像普通RDD一样处理实时数据

• Spark GraphX

  • 分布式图处理框架

  控制图、并行图操作和计算的一组算法和工具的集合。GraphX扩展了RDD API，包含控制图、创建子图、访问路径上所有顶点的操作

• Spark MLlib

  • 构建在Spark上的分布式机器学习库

  一个常用机器学习算法库，算法被实现为对RDD的Spark操作。这个库包含可扩展的学习算法，比如分类、回归等需要对大量数据集进行迭代的操作。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-0EEqGelX-1597058270299)(…/…/…/software/typora/img/image-20200804091326775.png)]

四、Spark环境部署

1、安装配置

• 选择Spark2.2版本，下载地址
  • https://archive.apache.org/dist/spark/spark-2.2.0/spark-2.2.0-bin-hadoop2.7.tgz
• 解压并配置环境变量
  • SPARK_HOME
• Spark配置文件
  • $SPARK_HOME/conf/spark-env.sh

export SPARK_MASTER_HOST=192.168.253.150
	//本机的ip地址
export SPARK_MASTER_PORT=7077
	
export SPARK_WORKER_CORES=2
	//2个核心 不能比分配的虚拟机核数大
export SPARK_WORKER_MEMORY=3g

export SPARK_MASTER_WEBUI_PORT=8888

• $SPARK_HOME/conf/slaves

localhost
    //由于是单机所以是本地

• 启动Spark
  • 先进入spark的sbin目录 ./start-all.sh启动Master和Worker任务
  • 进入bin目录./spark-shell进入shell界面

2、测试

• spark-shell：Spark自带的交互式工具
  • 本机
    • spark-shell --master local[*]
  • Standalone(需启动spark)
    • spark-shell --master spark://MASTERHOST:7077
  • YARN
    • spark-shell --master yarn

scala> sc.textFile("file:///opt/data/Spark/spark.txt").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).collect
res0: Array[(String, Int)] = Array((hive,1), ("",1), (hello,4), (world,1), (spark,1), (hadoop,1))

五、Spark架构设计

1、运行架构

• 在驱动程序中，通过SparkContext主导应用的执行
• SparkContext可以连接不同类型的Cluster Manager（Standalone、YARN、Mesos），连接后，获得集群节点上的Executor
• 一个Worker节点默认一个Executor，可通过SPARK_WORKER_INSTANCES调整
• 每个应用获取自己的Executor
• 每个Task处理一个RDD分区

2、运行基本流程

1、当一个spark应用被提交时，首先需要为这个应用构建起基本的运行环境，即由任务控制节点（Driver）创建一个SparkContext对象，由Sparkcontext负责和资源管理器的通信以及进行资源的申请、任务的分配和监控等，SparkContext会向资源管理器注册并申请运行Executor的资源，SparkContext可以看成是应用程序连接集群的通道。

2、资源管理器为Executor分配资源，并启动Executor进程，Executor运行情况将随着"心跳"发送到资源管理器上。

3、SparkContext根据RDD的依赖关系构建DAG图，DAG图提交给DAG调度器进行解析，将DAG图分解成多个‘阶段’（每个阶段都是一个任务集），并且计算出各个阶段之间的依赖关系，然后把一个个“任务集”提交给底层的任务调度器进行处理；Executor向SparkContext申请任务，任务调度器将任务分发给Executor运行，同时，SparkContext将应用程序代码发放给Executor。

4、任务在Executor上运行，把执行结果反馈给任务调度器，然后反馈给DAG调度器，运行完毕后写入数据并释放所有资源。+

3、核心组件

术语	说 明
Application	建立在Spark上的用户程序，包括Driver代码和运行在集群各节点Executor中的代码
Driver program	驱动程序。Application中的main函数并创建SparkContext
Cluster Manager	在集群（Standalone、Mesos、YARN）上获取资源的外部服务
Worker Node	集群中任何可以运行Application代码的节点
Executor	某个Application运行在worker节点上的一个进程
Job	包含多个Task组成的并行计算，往往由Spark Action触发生成，一个Application中往往会产生多个Job
Stage	每个Job会被拆分成多组Task，作为一个TaskSet，其名称为Stage
Task	被送到某个Executor上的工作单元

六、Spark API

1、SparkContext

• 连接Driver与Spark Cluster（Workers）
• Spark的主入口
• 每个JVM仅能有一个活跃的SparkContext
• SparkContext.getOrCreate

object SparkContextDemo extends App {
  //创建一个spark.context对象
  val conf: SparkConf = new SparkConf().setMaster("local[2]").setAppName("spark")
  val sc: SparkContext = SparkContext.getOrCreate(conf)
  sc.textFile("D:\\study files\\Spark\\test\\spark.txt")
    .flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_)
    .collect.foreach(println)
  sc.stop()
}

2、SparkSession

• Spark 2.0+应用程序的主入口：包含了SparkContext、SQLContext、HiveContext以及StreamingContext
• SparkSession.getOrCreate

object SparkSessionDemo extends App {
  //创建一个SparkSession对象
  val spark: SparkSession = SparkSession.builder()
    .master("local[2]").appName("SparkSessionTest").getOrCreate()
  val sc: SparkContext = spark.sparkContext
  sc.textFile("D:\\study files\\Spark\\test\\spark.txt")
    .flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_)
    .collect.foreach(println)
  sc.stop()
}

3、RDD核心

• RDD
  • Spark核心，主要数据抽象
• Dataset
  • 从Spark1.6开始引入的新的抽象，特定领域对象中的强类型集合，它可以使用函数或者相关操作并行地进行转换等操作
• DataFrame
  • DataFrame是特殊的Dataset

七、RDD

1、RDD概念

• RDD是弹性分布式数据集（Resilient Distributed Datasets）
  • 分布式（ Distributed）
    • 数据的计算并非只局限于单个节点，而是多个节点之间协同计算得到
  • 数据集（ Datasets）
    • RDD是只读的、分区记录的集合，每个分区分布在集群的不同节点上
    • RDD并不存储真正的数据，只是对数据和操作的描述
  • 弹性（Resilient）
    • 自动进行存储方式的切换，RDD优先存储内存中，内存不足将自动写入磁盘
    • 基于Linage的高效容错机制，在任何时候都能进行重算，根据数据血统，可以自动从节点失败中恢复分区，各个分片之间的数据互不影响
    • Stage失败自动重试 / Task失败自动重试
    • Checkpoint和Persist， checkpoint持久化到文件系统

2、五大特性

• 一系列的分区（分片）信息，每个任务处理一个分区
• 每个分区上都有compute函数，计算该分区中的数据
• RDD之间有一系列的依赖
• 分区器决定数据（key-value）分配至哪个分区
• 优先位置列表，将计算任务分派到其所在处理数据块的存储位置（移动数据不如移动计算）

3、RDD与DAG

• 两者是Spark提供的核心抽象
• DAG（有向无环图）反映了RDD之间的依赖关系

4、RDD编程流程

5、RDD的创建

（1）使用集合创建RDD

val rdd1:RDD[Int] =sc.parallelize(List(1,2,3,4,5,6,7,8))
println(rdd1.partitions.size)
val rdd2:RDD[Int] =sc.parallelize(List(1,2,3,4,5,6,7,8),6)
println(rdd2.partitions.size)
val rdd3: RDD[Int] = sc.makeRDD(List(1,2,3,4,5),10)
println(rdd3.partitions.size)
    //分区数量
    12
	6
	10

1、Spark默认会根据集群的情况来设置分区的数量，也可以通过parallelize()第二参数来指定
2、Spark会为每一个分区运行一个任务进行处理

（2）通过加载文件产生RDD

object createRDDDemo extends App {
  //创建一个spark.context对象
  val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("spark")
  val sc: SparkContext = SparkContext.getOrCreate(conf)
  val rdd1: RDD[String] = sc.textFile("file:///D:\\study files\\Spark\\project\\Spark_beijing\\src\\data\\hello.txt")

  rdd1.collect().foreach(println)
  sc.stop()

}

• 加载“file://……”时，以local运行仅需一份本地文件，以Spark集群方式运行，应保证每个节点均有该文件的本地副本

• 支持目录、压缩文件以及通配符

sc.textFile("/my/directory")
sc.textFile("/my/directory/*.txt")
sc.textFile("/my/directory/*.gz")

1、Spark默认访问HDFS
2、Spark默认为HDFS文件的每一个数据块创建一个分区，也可以通过textFile()第二个参数指定，但只能比数据块数量多

（3）创建PairRDD的方法

• SparkContext.wholeTextFiles()：可以针对一个目录中的大量小文件返回<filename,fileContent>作为PairRDD
  • 普通RDD：org.apache.spark.rdd.RDD[data_type]
  • PairRDD：org.apache.spark.rdd.RDD[(key_type,value_type)]

object CreatePairRDDDemo extends App {
  //创建一个spark context对象
  val conf = new SparkConf().setMaster("local[*]").setAppName("sparkTest6")
  val sc = SparkContext.getOrCreate(conf)

  val pairRDD =sc.wholeTextFiles("file:///D:\\study files\\Spark\\project\\Spark_beijing\\src\\data")
  pairRDD.foreach(println)
  sc.stop()
}

//输出结果

（4）其他创建RDD的方法

• SparkContext.sequenceFile[K,V]()
  • Hadoop SequenceFile的读写支持
• SparkContext.hadoopRDD()、newAPIHadoopRDD()
  • 从Hadoop接口API创建
• SparkContext.objectFile()
  • RDD.saveAsObjectFile()的逆操作

八、RDD操作

详情可参考官方中文文档：http://spark.apachecn.org/#/docs/4

1、RDD分区

• 分区是被拆分并发送到节点的RDD的不同块之一
  • 我们拥有的分区越多，得到的并行性就越强
  • 每个分区都是被分发到不同Worker Node的候选者
  • 每个分区对应一个Task
    

分区的作用：

一个HDFS文件的RDD将文件的每个文件块表示为一个分区，并且知道每个文件块的位置信息。这些对应着数据块的分区分布到集群的节点中，因此，分区的多少涉及对这个RDD进行并行计算的粒度。首先，分区是一个逻辑概念， 变换前后的新旧分区在物理上可能是同一块内存或者是存储。

需要注意的是，如果没有指定分区数将使用默认值，而默认值是该程序所分配到CPU核数，如果是从HDFS文件创建，默认为文件的数据块数。

2、RDD的操作算子

• 分为lazy与non-lazy两种
  • Transformation（lazy）：也称转换操作、转换算子
  • Actions（non-lazy）：立即执行，也称动作操作、动作算子

常用Transformation算子：

Transformation算子	含义
map(func)	返回一个新的RDD，该RDD由每一个输入元素经过func函数转换后组成
filter(func)	返回一个新的RDD，该RDD由经过func函数计算后返回值为true的输入元素组成
flatMap(func)	类似于map，但是每一个输入元素可以被映射为0或多个输出元素（所以func应该返回一个序列，而不是单一元素）
mapPartitions(func)	类似于map，但独立地在RDD的每一个分片上运行，因此在类型为T的RDD上运行时，func的函数类型必须是Iterator[T] => Iterator[U]
mapPartitionsWithIndex(func)	类似于mapPartitions，但func带有一个整数参数表示分片的索引值，因此在类型为T的RDD上运行时，func的函数类型必须是(Int, Interator[T]) => Iterator[U]
union(otherDataset)	对源RDD和参数RDD求并集后返回一个新的RDD
intersection(otherDataset)	对源RDD和参数RDD求交集后返回一个新的RDD
groupByKey([numTasks])	在一个(K,V)的RDD上调用，返回一个(K, Iterator[V])的RDD
reduceByKey(func, [numTasks])	在一个(K,V)的RDD上调用，返回一个(K,V)的RDD，使用指定的reduce函数，将相同key的值聚合到一起，与groupByKey类似，reduce任务的个数可以通过第二个可选的参数来设置
sortByKey([ascending], [numTasks])	在一个(K,V)的RDD上调用，K必须实现Ordered接口，返回一个按照key进行排序的(K,V)的RDD
sortBy(func,[ascending], [numTasks])	与sortByKey类似，但是更灵活
join(otherDataset, [numTasks])	在类型为(K,V)和(K,W)的RDD上调用，返回一个相同key对应的所有元素对在一起的(K,(V,W))的RDD
cogroup(otherDataset, [numTasks])	在类型为(K,V)和(K,W)的RDD上调用，返回一个(K,(Iterable,Iterable))类型的RDD
coalesce(numPartitions)	减少 RDD 的分区数到指定值。
repartition(numPartitions)	重新给 RDD 分区
repartitionAndSortWithinPartitions(partitioner)	重新给 RDD 分区，并且每个分区内以记录的 key 排序

常用Action算子

Action算子	含义
reduce(func)	reduce将RDD中元素前两个传给输入函数，产生一个新的return值，新产生的return值与RDD中下一个元素（第三个元素）组成两个元素，再被传给输入函数，直到最后只有一个值为止。
collect()	在驱动程序中，以数组的形式返回数据集的所有元素
count()	返回RDD的元素个数
first()	返回RDD的第一个元素（类似于take(1)）
take(n)	返回一个由数据集的前n个元素组成的数组
takeOrdered(n, [ordering])	返回自然顺序或者自定义顺序的前 n 个元素
saveAsTextFile(path)	将数据集的元素以textfile的形式保存到HDFS文件系统或者其他支持的文件系统，对于每个元素，Spark将会调用toString方法，将它装换为文件中的文本
saveAsSequenceFile(path)	将数据集中的元素以Hadoop sequencefile的格式保存到指定的目录下，可以使HDFS或者其他Hadoop支持的文件系统。
saveAsObjectFile(path)	将数据集的元素，以 Java 序列化的方式保存到指定的目录下
countByKey()	针对(K,V)类型的RDD，返回一个(K,Int)的map，表示每一个key对应的元素个数。
foreach(func)	在数据集的每一个元素上，运行函数func
foreachPartition(func)	在数据集的每一个分区上，运行函数func

3、RDD转换算子

• 对于转换操作，RDD的所有转换都不会直接计算结果
  • 仅记录作用于RDD上的操作
  • 当遇到动作算子（Action）时才会进行真正计算

（1）map算子

• 对RDD中的每个元素都执行一个指定的函数来产生一个新的RDD
• 任何原RDD中的元素在新RDD中都有且只有一个元素与之对应
• 输入分区与输出分区一一对应

object MapDemo extends App {
  //创建一个spark context对象
  val conf = new SparkConf().setMaster("local[*]").setAppName("sparkTest6")
  val sc = SparkContext.getOrCreate(conf)

  val a = sc.parallelize(1 to 5)
  val b = a.map(x=>x*2)
  a.collect().foreach(println)
  println("----------------------")
  b.collect().foreach(println)
  println("----------------------")
  //TODO:使用map将普通RDD转换成pairRDD
  val c = a.map(x=>(x,1))
  println(c.collect().mkString("\t"))
}

//输出结果
1
2
3
4
5
----------------------
2
4
6
8
10
----------------------
(1,1)	(2,1)	(3,1)	(4,1)	(5,1)

（2）filter算子

• 对元素进行过滤，对每个元素应用指定函数，返回值为true的元素保留在新的RDD中

object FilterRDD extends App {
  //创建一个spark context对象
  val conf = new SparkConf().setMaster("local[*]").setAppName("sparkTest6")
  val sc = SparkContext.getOrCreate(conf)

  val a = sc.parallelize(1 to 5)
  val b = a.filter(x=>x%2==0)
  println(b.collect().mkString("\t"))

}

//输出结果
2	4

（3）mapValue算子

• 原RDD中的Key保持不变，与新的Value一起组成新的RDD中的元素，仅适用于PairRDD

object MapValueDemo extends App {
  //创建一个spark context对象
  val conf = new SparkConf().setMaster("local[*]").setAppName("sparkTest6")
  val sc = SparkContext.getOrCreate(conf)
  val a:RDD[String] = sc.parallelize(List("dog","cat","panda"))

  //TODO:使用map将普通RDD转换成pairRDD
  val pairRDD: RDD[(Int, String)] = a.map(x=>(x.length,x))
  println(pairRDD.collect().mkString("\t"))
  //TODO:mapValues对pairRDD的value进行处理
  val d: RDD[(Int, String)] = pairRDD.mapValues(x=>x+" say:"+"hungry")
  println(d.collect().mkString("\t"))
}

//输出结果
(3,dog say:hungry)	(3,cat say:hungry)	(5,panda say:hungry)

(4)FlatMap算子

• 类似于map，但是每一个输入元素可以被映射为0或多个输出元素

object FlatMapDemo extends App {
  //创建一个spark context对象
  val conf = new SparkConf().setMaster("local[*]").setAppName("sparkTest6")
  val sc = SparkContext.getOrCreate(conf)

  private val rdd1: RDD[List[Int]] = sc.parallelize(List(List(1,2,3),List(4,5,6),List(7,8,9)))
  private val rdd2: RDD[Char] = rdd1.flatMap(x=>x.mkString(""))
  println(rdd2.collect().mkString(""))
}
//输出结果
123456789

（5）MapPartitionsWithIndex算子

object MapPartitionsWithIndexDemo extends App {
  //创建一个spark context对象
  val conf = new SparkConf().setMaster("local[5]").setAppName("sparkTest6")
  val sc = SparkContext.getOrCreate(conf)

  val a = sc.parallelize(1 to 5)

  //List(1,2)
  val rdd: RDD[(Int, Int)] = a.mapPartitionsWithIndex((index, items)=>(items.map(x=>(index,x))))
  println(rdd.collect().mkString("\t"))

}

//输出结果
(0,1)	(1,2)	(2,3)	(3,4)	(4,5)

（6）更多的转换算子

• distinct
• reduceByKey

object ReduceByKeyDemo extends App {
  //创建一个spark context对象
  val conf = new SparkConf().setMaster("local[5]").setAppName("sparkTest6")
  val sc = SparkContext.getOrCreate(conf)

  val rdd1 = sc.parallelize(List("dog","cat","pig","salmon"))
  private val rdd2: RDD[(Int, String)] = rdd1.map(x=>(x.length,x))
  println(rdd2.collect().mkString(" "))
  //TODO:reduceByKey
  private val rdd3: RDD[(Int, String)] = rdd2.reduceByKey((a, b)=>a+b)
  println(rdd3.collect().mkString(" "))
  //TODO:groupByKey
  private val rdd4: RDD[(Int, Iterable[String])] = rdd2.groupByKey()
  println(rdd4.collect().mkString(" "))
  //TODO:sortByKey
  println("----------------------------------------")
  println(rdd2.sortByKey().collect().mkString(" "))
  println(rdd2.sortByKey(false).collect().mkString(" "))
}

//输出结果
(3,dog) (3,cat) (3,pig) (6,salmon)
(6,salmon) (3,dogcatpig)
(6,CompactBuffer(salmon)) (3,CompactBuffer(dog, cat, pig))
----------------------------------------
(3,dog) (3,cat) (3,pig) (6,salmon)
(6,salmon) (3,dog) (3,cat) (3,pig)

• groupByKey
• sortByKey

object SortByDemo extends App {
  //创建一个spark context对象
  val conf = new SparkConf().setMaster("local[5]").setAppName("sparkTest6")
  val sc = SparkContext.getOrCreate(conf)

  val a = sc.parallelize(List(10,2,3,5,13,6,3))
  println(a.sortBy(x => x).collect().mkString(" "))
  println(a.sortBy(x => x * (-1)).collect().mkString(" "))
  
}

//输出结果
2 3 3 5 6 10 13
13 10 6 5 3 3 2

• union

object UnionDemo extends App {
  //创建一个spark context对象
  val conf = new SparkConf().setMaster("local[5]").setAppName("sparkTest6")
  val sc = SparkContext.getOrCreate(conf)

  val a = sc.parallelize(1 to 10)
  val b = sc.parallelize(4 to 15)
  val unionRDD: RDD[Int] = a.union(b)
  println(unionRDD.collect().mkString(" "))
}
//输出结果
1 2 3 4 5 6 7 8 9 10 4 5 6 7 8 9 10 11 12 13 14 15

• join和cogroup

object JoinDemo extends App {
  //创建一个spark context对象
  val conf = new SparkConf().setMaster("local[5]").setAppName("sparkTest6")
  val sc = SparkContext.getOrCreate(conf)

  //创建两个RDD
  val rdd1 = sc.parallelize(Array((1,"a"),(2,"b"),(3,"c")))
  val rdd2 = sc.parallelize(Array((1,4),(2,5),(3,6)))

  //TODO:进行join操作 在类型为（K,V）和（K,W）的RDD上调用
  //TODO：返回一个形同额key对应的所有元素在一起的（K,(V,W)）的调用
  val rdd3 = rdd1.join(rdd2)
  println(rdd3.collect().mkString(" "))
}

//输出结果
(1,(a,4)) (2,(b,5)) (3,(c,6))

object CogroupDemo extends App {
  //创建一个spark context对象
  val conf = new SparkConf().setMaster("local[5]").setAppName("sparkTest6")
  val sc = SparkContext.getOrCreate(conf)

  //创建两个RDD
  val rdd1 = sc.parallelize(Array((1,"a"),(2,"b"),(3,"c")))
  val rdd2 = sc.parallelize(Array((1,4),(2,5),(3,6)))

  //TODO:进行cogroup操作 在类型为（K,V）和（K,W）的RDD上调用
  //TODO：返回一个形同额key对应的所有元素在一起的（K,(V,W)）的调用
  val rdd3 = rdd1.cogroup(rdd2)
  println(rdd3.collect().mkString(" "))

}

//输出结果
(1,(CompactBuffer(a),CompactBuffer(4))) (2,(CompactBuffer(b),CompactBuffer(5))) (3,(CompactBuffer(c),CompactBuffer(6)))

• count
• sample

object SampleDemo extends App {
  //创建一个spark context对象
  val conf = new SparkConf().setMaster("local[5]").setAppName("sparkTest6")
  val sc = SparkContext.getOrCreate(conf)

  val a = sc.parallelize(1 to 10)
  a.sample(true,0.4,2).collect().foreach(println)
}
//输出结果
1
2
2
6
6
7
8
9
9
10

• repartition和coalesce

object RePartitionDemo extends App {
  //创建一个spark context对象
  val conf = new SparkConf().setMaster("local[5]").setAppName("sparkTest6")
  val sc = SparkContext.getOrCreate(conf)

  val a:RDD[Int]  = sc.parallelize(1 to 10,4)
  //TODO:  repartition重新分区
  private val rdd: RDD[Int] = a.repartition(2)
  println(a.partitions.size)
  println(rdd.partitions.size)
  //TODO:  coalesce 减少到2个分区 可以指定是否发生shuffle
  private val rdd1: RDD[Int] = a.coalesce(2)
  println(rdd1.partitions.size)
}

//输出结果
4
2
2

同时也可以这样改变分区：

object PartitionDemo extends App {
  //创建一个spark context对象
  val conf = new SparkConf().setMaster("local[5]").setAppName("sparkTest6")
  val sc = SparkContext.getOrCreate(conf)

  //创建pairRDD
  val rdd1 = sc.parallelize(Array((1,"a"),(2,"b"),(3,"c")))
  //TODO:进行partition操作 对pariRdd进行分区操作
  //TODO：如果原有的pairRDD和现有的partionRDD是一致的话就不进行分区
  //TODO:否则会生成shufflerdd会产生shuffle过程
  val rdd2=rdd1.partitionBy(new org.apache.spark.HashPartitioner(2))
  println(rdd2.partitions.size)
  sc.stop()
}

//输出结果
2

• zip

object ZipDemo extends App {
  //创建一个spark context对象
  val conf = new SparkConf().setMaster("local[5]").setAppName("sparkTest6")
  val sc = SparkContext.getOrCreate(conf)

  //创建两个rDD
  val rdd1 = sc.parallelize(Array(1,2,3,4,5))
  val rdd2 = sc.parallelize(Array("A","B","C","D","E"))
  //todo:将两个RDD组合成keyvalue的形式的rdd
  //todo：这里需要两个数量相同否则会跑出异常
  val rdd3 = rdd1.zip(rdd2)
  println(rdd3.collect().mkString(" "))
  
  //元素数量不同会报错
  val rdd4 = sc.parallelize(Array("A","B","C","D"))
  val rdd5 = rdd1.zip(rdd4)
  println(rdd5.collect().mkString(" "))

}

4、RDD动作算子

• 本质上动作算子通过SparkContext执行提交作业操作，触发RDD DAG（有向无环图）的执行
• 所有的动作算子都是急迫型（non-lazy），RDD遇到Action就会立即计算

(1)count

返回的是数据集中的元素的个数

val rdd=sc.parallelize(List(1,2,3,4,5,6))
rdd.count
//6

(2)collect

以Array返回RDD的所有元素。一般在过滤或者处理足够小的结果的时候使用

val rdd=sc.parallelize(List(1,2,3,4,5,6))
println(rdd.collect.mkString(" "))
// 1 2 3 4 5 6

(3)take

返回前n个元素

val rdd=sc.parallelize(List(1,2,3,4,5,6))
println(rdd.take(3).mkString(" "))
    //1 2 3

(4)first

返回RDD第一个元素

val rdd=sc.parallelize(List(1,2,3,4,5,6))
println(rdd.first())
    //1

(5)reduce

根据指定函数，对RDD中的元素进行两两计算，返回计算结果

val a=sc.parallelize(1 to 100)
a.reduce((x,y)=>x+y)
a.reduce(_+_)		//与上面等价
val b=sc.parallelize(Array(("A",0), ("A",2), ("B",1), ("B",2), ("C",1)))
b.reduce((x,y)=>{(x._1+y._1,x._2+y._2)})		//(AABBC,6)

(6)foreach

对RDD中的每个元素都使用指定函数，无返回值

val rdd=sc.parallelize(1 to 100)
rdd.foreach(println)

(7)lookup

用于PairRDD,返回K对应的所有V值

val rdd=sc.parallelize(List(('a',1), ('a',2), ('b',3), ('c',4)))
println(rdd.lookup('a'))
    //WrappedArray(1, 2)

(8)最值

max、min返回最大值、最小值

val c:RDD[Int] = sc.parallelize(1 to 50)
println(c.max())
println(c.min())
    //50
	//1

(9)saveAsTextFile

保存RDD数据至文件系统

val d:RDD[Int] = sc.parallelize(1 to 10)
d.saveAsTextFile("src/data/test")

定函数，对RDD中的元素进行两两计算，返回计算结果

val a=sc.parallelize(1 to 100)
a.reduce((x,y)=>x+y)
a.reduce(_+_)		//与上面等价
val b=sc.parallelize(Array(("A",0), ("A",2), ("B",1), ("B",2), ("C",1)))
b.reduce((x,y)=>{(x._1+y._1,x._2+y._2)})		//(AABBC,6)

(6)foreach

对RDD中的每个元素都使用指定函数，无返回值

val rdd=sc.parallelize(1 to 100)
rdd.foreach(println)

(7)lookup

用于PairRDD,返回K对应的所有V值

val rdd=sc.parallelize(List(('a',1), ('a',2), ('b',3), ('c',4)))
println(rdd.lookup('a'))
    //WrappedArray(1, 2)

(8)最值

max、min返回最大值、最小值

val c:RDD[Int] = sc.parallelize(1 to 50)
println(c.max())
println(c.min())
    //50
	//1

(9)saveAsTextFile

保存RDD数据至文件系统

val d:RDD[Int] = sc.parallelize(1 to 10)
d.saveAsTextFile("src/data/test")