什么是Spark

Apache Spark是专为大规模数据处理而设计的快速通用的计算引擎 [1] 。现在形成一个高速发展应用广泛的生态系统。

Spark优势

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

Spark的技术栈

1.Spark Core
核心组件，分布式计算引擎。其实现了 Spark 的作业调度、内存管理、容错、
与存储系统交互等基本功能，并针对弹性分布式数据集（RDD）提供了丰富的操
作。
2.Spark SQL
又一个高性能的基于 Hadoop 的 SQL 解决方案。部分用法与 Hive 非常类似。

3.Spark Streaming
基于 Spark Core 实现的高吞吐量、具备容错机制的准实时流处理系统。将流
式计算分解成一系列小批处理作业，也称微批处理。
4.Spark GraphX
分布式图处理框架，支持图并行计算。现在已经提供了很多算法，新的算法
还在不断加入。
5.Spark MLlib
构建在 Spark 上的分布式机器学习库。是 Spark 对常用的机器学习算法的实
现库，还提供了相关的测试与数据生成器。

Spark shell

• 1.本机
  spark-shell --master local[*]

• Standalone(需启动spark)
  spark-shell --master spark://MASTERHOST:7077

• YARN
  spark-shell --master yarn

Spark运行架构

• 在驱动程序中，通过SparkContext主导应用的执行

• SparkContext可以连接不同类型的Cluster
  Manager（Standalone、YARN、Mesos），连接后，获得集群节点上的Executor

• 一个Worker节点默认一个Executor，可通过SPARK_WORKER_INSTANCES调整

• 每个应用获取自己的Executor

• 每个Task处理一个RDD分区

Spark架构核心组件

Spark API

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

import org.apache.spark.{SparkConf, SparkContext}val conf=new SparkConf().setMaster("local[2]").setAppName("HelloSpark")
val sc=SparkContext.getOrCreate(conf)

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

import org.apache.spark.sql.SparkSession
val spark = SparkSession.builder.master("local[2]").appName("appName").getOrCreate()

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

RDD

4. 简单的解释
   RDD是将数据项拆分为多个分区的集合，存储在集群的工作节点上的内存和磁盘中，并执行正确的操作
5. 复杂的解释
   RDD是用于数据转换的接口,如map、filter等
   RDD指向了存储在HDFS、Cassandra、HBase等、或缓存（内存、内存+磁盘、仅磁盘等）、或在故障或缓存收回时重新计算其他RDD分区中的数据
6. RDD是弹性分布式数据集
   1.分布式（ Distributed）
   数据的计算并非只局限于单个节点，而是多个节点之间协同计算得到
   2.数据集（ Datasets）
   RDD是只读的、分区记录的集合，每个分区分布在集群的不同节点上
   RDD并不存储真正的数据，只是对数据和操作的描述
   3.弹性（Resilient）
   自动进行存储方式的切换，RDD优先存储内存中，内存不足将自动写入磁盘
   基于Linage的高效容错机制，在任何时候都能进行重算，根据数据血统，可以自动从节点失败中恢复分区，各个分片之间的数据互不影响
   Stage失败自动重试 / Task失败自动重试
   Checkpoint和Persist， checkpoint持久化到文件系统
7. RDD的五大特性
   1.一系列的分区（分片）信息，每个任务处理一个分区
   2.每个分区上都有compute函数，计算该分区中的数据
   3.RDD之间有一系列的依赖
   4.分区器决定数据（key-value）分配至哪个分区
   5.优先位置列表，将计算任务分派到其所在处理数据块的存储位置（移动数据不如移动计算）
   

创建RDD

1. 使用集合创建RDD

val rdd=sc.parallelize(List(1,2,3,4,5,6))
rdd.count
rdd.partitions.size
//设置分区数
val rdd=sc.parallelize(List(1,2,3,4,5,6),5)
rdd.partitions.size
val rdd=sc.makeRDD(List(1,2,3,4,5,6))
//Spark默认会根据集群的情况来设置分区的数量，也可以通过parallelize()第二参数来指定
//Spark会为每一个分区运行一个任务进行处理

2. 通过加载文件产生RDD

val distFile=sc.textFile("file:///home/hadoop/data/hello.txt")
distFile.count
val distHDFSFile=sc.textFile("hdfs://hadoop000:9000/hello.txt")
//加载“file://……”时，以local运行仅需一份本地文件，以Spark集群方式运行，应保证每个节点均有该文件的本地副本

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

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

3. 创建PairRDD的方法
   SparkContext.wholeTextFiles()：可以针对一个目录中的大量小文件返回<filename,fileContent>作为PairRDD
   Spark 为包含键值对类型的 RDD 提供了一些专有的操作，比如：reduceByKey()、groupByKey()……
   也可以通过键值对集合创建PairRDD：
   sc.parallelize(List((1,2),(1,3)))

RDD转换算子

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

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

//map把普通RDD变成PairRDDval conf: SparkConf = new SparkConf().setMaster("local[4]").setAppName("sparkTest6")val sc: SparkContext = SparkContext.getOrCreate(conf)private val rdd: RDD[Int] = sc.parallelize(1 to 10)private val a: RDD[(Int, Int)] = rdd.map((_,1))println(a.collect().mkString(" "))

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

	val conf: SparkConf = new SparkConf().setMaster("local[4]").setAppName("sparkTest6")val sc: SparkContext = SparkContext.getOrCreate(conf)private val rdd: RDD[Int] = sc.parallelize(1 to 10)private val a: RDD[Int] = rdd.filter(x=>x%2==0)println(a.collect().mkString(" "))

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

 val conf: SparkConf = new SparkConf().setMaster("local[2]").setAppName("sparkTest6")val sc: SparkContext = SparkContext.getOrCreate(conf)private val rdd: RDD[String] = sc.parallelize(List("lion","tiger","panda","dog","cat"))private val a: RDD[(Int, String)] = rdd.map(x=>(x.length,x))a.collect().foreach(println)private val b: RDD[(Int, String)] = a.mapValues(x=>x+"hhhhh")b.collect().foreach(println)

4. mapPartitions

//TODO:mapPartitions每次处理一个分区的数据
//所以
val conf: SparkConf = new SparkConf().setMaster("local[6]").setAppName("sparkTest6")val sc: SparkContext = SparkContext.getOrCreate(conf)private val rdd: RDD[Int] = sc.parallelize(1 to 10)private val a: RDD[Int] = rdd.mapPartitions(x=>x.map(_*2))println(a.collect().mkString(" "))

5. mapPartitionsWithIndex()

//mapPartitionsWithIndex()传入的方法需要两个参数，一个为分区Id，另一个为分区数据，该方法可用来查看各分区的数据
val conf: SparkConf = new SparkConf().setMaster("local[6]").setAppName("sparkTest6")val sc: SparkContext = SparkContext.getOrCreate(conf)private val rdd: RDD[Int] = sc.parallelize(1 to 10)private val a: RDD[(Int, Int)] = rdd.mapPartitionsWithIndex((index, items)=>(items.map(x=>(index,x))))a.collect().foreach(println)

6. flatMap

val conf: SparkConf = new SparkConf().setMaster("local[6]").setAppName("sparkTest6")val sc: SparkContext = 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(" "))

7. reduceByKey和groupByKey

//reducebyKey会先在本地机器上进行局部聚合，然后在移动数据，进行全局聚合
// groupbyKey会先将数据进行移动，再做聚合
val conf: SparkConf = new SparkConf().setMaster("local[6]").setAppName("sparkTest6")val sc: SparkContext = SparkContext.getOrCreate(conf)//TODO:reduceByKeyval rdd1 =sc.parallelize(List("dog","cat","pigg","salmon"))private val rdd2: RDD[(Int, String)] = rdd1.map(x=>(x.length,x))println(rdd2.collect().mkString(" "))private val rdd3: RDD[(Int, String)] = rdd2.reduceByKey(_+_)println(rdd3.collect().mkString(" "))//TODO:groupByKeyprivate val rdd4: RDD[(Int, Iterable[String])] = rdd2.groupByKey()println(rdd4.collect().mkString(" "))

8. repartition
   随机地重新 Shuffle RDD 中的数据，以创建更多或更
   少的分区，并在它们之间进行平衡。

  val conf: SparkConf = new SparkConf().setMaster("local[6]").setAppName("sparkTest6")val sc: SparkContext = SparkContext.getOrCreate(conf)val a: RDD[Int] = sc.parallelize(1 to 10,4)private val rdd2: RDD[Int] = a.repartition(6)println(a.partitions.size)println(rdd2.partitions.size)private val rdd3: RDD[Int] = a.coalesce(6,true)println(rdd3.partitions.size)

9. sample

//sample(withReplacement:Boolean,fraction:Double,seed:Long)
//true代表如果抽中A元素,之后还可以抽取A元素
//false代表抽住了A元素,之后都不在抽取A元素
//fraction:抽样的比例
//seed:抽样算法的初始值
val conf: SparkConf = new SparkConf().setMaster("local[6]").setAppName("sparkTest6")val sc: SparkContext = SparkContext.getOrCreate(conf)val a: RDD[Int] = sc.parallelize(1 to 10)// TODO samplea.sample(false,0.4,2).collect().foreach(println)

11. sort by

 val conf: SparkConf = new SparkConf().setMaster("local[6]").setAppName("sparkTest6")val sc: SparkContext = SparkContext.getOrCreate(conf)val a: RDD[Int] = sc.parallelize(List(10,2,3,5,13,6,3))//TODO: sortBy 进行排序println(a.sortBy(x => x).collect().mkString(" "))println(a.sortBy(x => x * (-1)).collect().mkString(" "))

12. union

val conf: SparkConf = new SparkConf().setMaster("local[6]").setAppName("sparkTest6")val sc: SparkContext = SparkContext.getOrCreate(conf)val a: RDD[Int] = sc.parallelize(1 to 10)private val b: RDD[Int] = sc.parallelize(4 to 15)private val c: RDD[String] = sc.parallelize(List("hello","world"))
// TODO union  不去重private val unionRDD: RDD[Int] = a.union(b)println(unionRDD.collect().mkString(" "))println("----------我是分割线-----------")

13. intersection

  //创建一个spark context对象val conf: SparkConf = new SparkConf().setMaster("local[6]").setAppName("sparkTest6")val sc: SparkContext = SparkContext.getOrCreate(conf)val a: RDD[Int] = sc.parallelize(1 to 10)private val b: RDD[Int] = sc.parallelize(4 to 15)//TODO: intersection  交集
private val intersectionRDD: RDD[Int] = a.intersection(b)println(intersectionRDD.collect().mkString(" "))

14. distinct

  //创建一个spark context对象val conf: SparkConf = new SparkConf().setMaster("local[6]").setAppName("sparkTest6")val sc: SparkContext = SparkContext.getOrCreate(conf)val a: RDD[Int] = sc.parallelize(1 to 10)private val b: RDD[Int] = sc.parallelize(4 to 15)private val c: RDD[String] = sc.parallelize(List("hello","world"))
// TODO union  不去重private val unionRDD: RDD[Int] = a.union(b)
println(unionRDD.distinct().collect().mkString(" "))println(unionRDD.distinct(3).collect().mkString(" "))sc.stop()

行动算子

1. reduce

 val conf: SparkConf = new SparkConf().setMaster("local[6]").setAppName("sparkTest6")val sc: SparkContext = SparkContext.getOrCreate(conf)val a: RDD[Int] = sc.parallelize(1 to 5)//TODO: map 对前一个rdd里面传过来的数据，每一个元素做一个操作//private val b: RDD[Int] = a.map(x=>x*2)val b: RDD[Int] = a.map(_*2)//TODO: reduce 操作 对所有元素做聚合操作private val result: Int = b.reduce((x, y)=>x*y)println(result)

2. collect

  //TODO: count是统计元素数量println(a.count())

3. take

  //TODO:take 返回前n个元素private val arr2: Array[Int] = a.take(3)println(arr2.mkString(" "))

4. first

  //TODO:first返回第一个结果println("first返回的结果："+b.first())

5. lookup

  //TODO:lookup,求出相应key对应的所有valueprivate val result: Seq[String] = rdd2.lookup(3)println(result.mkString("-"))

6. saveAsTextFile
   将数据集的元素作为文本文件(或文本文件集)写
   入本地文件系统、HDFS 或任何其他 hadoop 支持
   的文件系统的给定目录中。Spark 将对每个元素调
   用 toString，将其转换为文件中的一行文本。

  val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("sparkTest6")val sc: SparkContext = SparkContext.getOrCreate(conf)val a: RDD[Int] = sc.parallelize(1 to 100,10)a.saveAsTextFile("file:///D:\\project\\2020\\Scala\\spark_day0804\\src\\data\\saveAsText10")