1. Spark执行流程
知识补充：RDD的依赖关系
RDD的依赖关系分为两类：窄依赖（Narrow Dependency）和宽依赖(Shuffle Dependency)
（1）窄依赖
窄依赖指的是父RDD中的一个分区最多只会被子RDD中的一个分区使用，意味着父RDD的一个分区内的数据是不能被分割的，子RDD的任务可以跟父RDD在同一个Executor一起执行，不需要经过Shuffle阶段去重组数据
窄依赖关系划分为两种：一对一依赖（OneToOneDependency）和范围依赖（RangeDependency）
- 一对一依赖
- 范围依赖
（2）宽依赖
指的是父RDD中的分区可能被多个子RDD分区使用。因为父RDD中一个分区内的数据会被分割，发送给子RDD的多个分区，因此宽依赖也意味着父RDD与子RDD之间存在着Shuffle过程
宽依赖只有一种：Shuffle依赖（ShuffleDependency）
什么是Shuffle：
父RDD的一个分区的数据，要给子RDD的多个分区，shuffle要有网络传输，但是有网络传输的，不一定就是shuflle
窄依赖每个 child RDD 的 partition 的生成操作都是可以并行的，而宽依赖则需要所有的 parent RDD partition shuffle 结果得到后再进行。
以join算子为例
父RDD一个分区中的数据，被分割发送给子RDD的不同分区，所以是宽依赖
特殊情况（同理其他算子）
两个RDD使用相同的分区器，事先已经分完组或分好区了，在调用join，使用相同的分区，并且没有改变RDD的分区数量，那就就是窄依赖
3个stage
1.1 提交任务
spark-submit --master spark://feng05:7070 --executor-memory 1g --total-executor-cores 4 --class cn.51doit.spark.WordCount /root/wc.jar hdfs://feng05:9000/wc hdfs:/feng05:9000/out0
spark任务执行模式
- Client模式(默认)：Driver是在SparkSubmit进程中，是在客户端
- Cluster模式：Driver是在集群中，不在SparkSubmit进程中
1.2 创建SparkContext
使用spark-submit脚本，会启动SparkSubmit进程，然后通过反射调用我们通过--class传入类的main方法。在main方法中，就是我们写的业务逻辑了，先创建SparkContext，向Master申请资源，然后Master跟worker通信，启动executor，然后所有的Executor向Driver反向注册。
1.3 创建RDD并构建DAG
DAG(Directed Acyclic Graph)叫做有向无环图，是一系列RDD转换关系的描述，原始的RDD通过一系列的转化就形成了DAG，然后根据RDD的依赖关系的不同将DAG划分为不同的stage。对于窄依赖，partition的转换处理在Stage中完成计算，没有stage的划分；对于宽依赖，由于有shuffle的存在，只能在parent RDD处理完后，才能开始接下来的计算，会有stage的划分，因此宽依赖是划分Stage的依据。
1.4 切分Stage，生成Task和TaskSet
1.5 将task序列化，调度到Executor中
1.6 executor将task反序列化，得到task，并在线程池中执行这个任务
2. Repartition和coalesce算子的区别
这两个算子都是用于重新分区的，Repartition底层调用的是coalesce，具体减员吗，如下
repartition源码
可见其底层调用的是coalesce，传入的shuffle系数为true
coalesce方法源码
可见当shuffle参数为true时，创建的才是ShuffledRDD，当shuffle参数为false时，创建的是CoalescedRDD
由此不难得出结论：repartition间的父RDD和子RDD一定是宽依赖，Coalesce则不一定，视调用此方法的shuffle定。
3.触发多次actions，速度不一样
触发多次actions时，后触发的action会比前面触发的action快很多，这是为什么？
第一次shuffle时，结果会被溢写进磁盘（由blockmanger管理），后面可以复用这个结果
以上是第二次进行action的操作，左边灰色表示的是之前action操作时，相关数据被溢写入磁盘中，此处就直接复用这些结果。所以会很快。
4. RDD的的深入理解
4.1 重要总结
在刚开始学习Spark时，为了方便理解，可以把Spark的RDD就当成Scala的一个普通的集合使用，Scala集合的方法和RDD上的方法很多功能是类似（底层实现不一样）的，比如map、flatMap、filter、reduce等，但是Scala的集合是本地的集合，而RDD是一个抽象、分布式的集合，RDD可以实现分布在多台机器上数据的计算
RDD本身不装真正要计算的数据，RDD里面装的是数据的描述信息，描述了以后从哪里读取数据，对RDD进行哪些操作（调用了什么方法，传入了（）什么函数），一旦触发Action，就会形成一个完整的DAG
在spark中，提交的应应⽤程序叫Application，⼀个Application中触发⼀次Action就提交⼀个Job（DAG），⼀个Job可以划分成⼀到多个Stage，⼀个Stage会⽣产多个Task（有⼏个分区就有⼏个Task），Task是Spark中最⼩的任务执⾏单元，在⽤⼀个Stage中，task的计算逻辑是⼀样的，只不过是计算的数据不⼀样。
Task是什么东西？Task就是⼀个java对象（实例），java对象中有属性和⽅法，属性：记录的描述信息（⽐如从哪⾥读取数据，读取哪个⽂件等）。⽅法：具体怎么计算（调⽤哪个算⼦、传⼊了什么函数）
Task是⼀个最小的执⾏单元，Task这个类不能我们⽤户⾃⼰实现，在Spark中，是根据RDD的转换关系（调⽤了哪个算⼦、传⼊了什么函数）⾃动⽣成的Task，这样⾮常的灵活
4.2 源码了解RDD是如何获取数据的（尝试，以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] = withScope {
val cleanF = sc.clean(f) //检测传入的函数能否被序列化，若函数中传入一个不能被序列化的引用数据类型就会报错（闭包单词：closure，进一步看源码会涉及）
new MapPartitionsRDD[U, T](this, (context, pid, iter) => iter.map(cleanF))
}
从此段代码可得到一个重要的信息：对RDD进行操作，本质上就是对每个分区进行操作
MapPartitionsRDD（部分）
当进一步点进iterator中去，会发现其返回的还是一个迭代器，源码如下
总结： 当触发Action算子时，ShuffleMapTask会进行shuffle准备，将数据通过shufflewrite写入磁盘，其过程为子RDD通过调用父RDD的迭代器获取父RDD的数据，父RDD又通过它的父RDD的迭代器获取父RDD的数据，以此类推，直到HadoopRDD。那么，Hadoop如何从HDFS获取数据的呢？通过网络迭代器从HDFS中拉取数据
4.3 RDD的典型错误例子
object BadStyle {
def main(args: Array[String]): Unit = {
//模式提交任务使用的是Client模式，SparkSubmit进程中包含SparkContext
val conf = new SparkConf().setAppName("BadStyle").setMaster("local[*]")
//Driver
val sc = new SparkContext(conf)
//Driver
val lines: RDD[String] = sc.textFile("/Users/star/Desktop/a.txt")
//Driver
val words: RDD[String] = lines.flatMap(_.split(" "))
//RDD是在Driver端创建的抽象集合
//调用的map方法也是在Driver端调用的
val result = words.map(w => {
//函数式在Executor中被执行的
//lines是RDD，RDD不能再Executor中
lines.map(l => w + l)
})
result.saveAsTextFile("/Users/star/Desktop/c")
println("Driver ############# ")
sc.stop()
println("Driver ############# 任务退出了")
}
}
RDD是在Driver端生成的，其不能在executor中，所以会报错，报错信息显示也是如此
改成如下就正确了
// 此代码在函数外
val count = lines.count()
// 此代码在函数内
w+count
5. Spark任务执行的经典问题
（1）SparkContext哪一端生成的？
Driver端（Driver是一个统称，DAGSchedule、TaskScheduler、BlockManager、ShuffleManager、BroadcastManager）
（2）DAG是在哪一端被构建的？
Driver端
（3）RDD是在哪一端生成的？
Driver端
（4）调用RDD的算子（Transformation和Action）是在哪一端调用的
Driver端
（5）RDD在调用Transformation和Action时需要传入一个函数，函数是在哪一端声明【定义】和传入的?
Driver端
（6）RDD在调用Transformation和Action时需要传入函数，请问传入的函数是在哪一端执行了函数的业务逻辑？
Executor中的Task指定的
（7）Task是在哪一端生成的呢？
Driver端,Task分为ShuffleMapTask和ResultTask
（8）DAG是在哪一端构建好的并被切分成一到多个Stage的
Driver
（9）DAG是哪个类完成的切分Stage的功能？
DAGScheduler
（10）DAGScheduler将切分好的Task以什么样的形式给TaskScheduler
TaskSet
（11）13.自定义的分区器这个类是在哪一端实例化的？
Driver端
（12）分区器中的getParitition方法在哪一端调用的呢？
Executror中的Task
（13）广播变量是在哪一端调用的方法进行广播的？
Driver端
（14）要广播的数据应该在哪一端先创建好再广播呢？
Driver端
（15）广播变量以后能修改吗？
不能修改
（16）广播变量广播到Executor后，一个Executor进程中有几份广播变量的数据
一份全部的广播的数据
6. 购物的相关计算
日志数据样例（以下截取的是其中的一条数据（一行，但便于观看就改成如下形式））：
约定
5.1 小程序的PV，UV，用户的区域分布
ShopKpi（此处没有用bean来封装日志文件的各属性）
package com._51doit.spark06
import com.alibaba.fastjson.{JSON, JSONException, JSONObject}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
object ShopKpi {
def main(args: Array[String]): Unit = {
//模式提交任务使用的是Client模式，SparkSubmit进程中包含SparkContext
val conf = new SparkConf().setAppName("ShopKpi").setMaster("local[*]")
val sc = new SparkContext(conf)
val lines: RDD[String] = sc.textFile("E:/javafile/spark/access-new.log")
// 计算pv（访问量）
val pv: Long = lines.count()
// 处理数据
val uidAndProvince: RDD[(String, String)] = lines.map(line => {
var openid: String = null
var province: String = null
var city: String = null
try {
val jsonObject: JSONObject = JSON.parseObject(line)
openid = jsonObject.getString("openid")
province = jsonObject.getString("province")
city = jsonObject.getString("city")
} catch {
case e: JSONException => {
// 处理错误的数据
}
}
if("北京市".equals(city)){
(openid, province)
} else{
(openid, province+city)
}
})
//过滤有问题的数据
val filteredRDD: RDD[(String, String)] = uidAndProvince.filter(t => t._1 != null && !t._1.equals(""))
filteredRDD.cache()
// 日活
val uv: Long = filteredRDD.keys.distinct().count()
// 用户的区域分布
val reduced: RDD[(String, Int)] = filteredRDD.distinct().map(t => (t._2, 1)).reduceByKey(_+_)
println(reduced.collect().toBuffer)
sc.stop()
}
}
补充
Json解析数据的两种形式
第一种（数据不封装到类中，用的时候直接获取）：
val jsonObject: JSONObject = JSON.parseObject(line)
openid = jsonObject.getString("openid")
province = jsonObject.getString("province")
第二种（数据封装到bean中，一般使用这这种）
val logBeanRDD: RDD[LogBeanV2] = lines.map(line => {
var logBean: LogBeanV2 = null
try {
logBean = JSON.parseObject(line, classOf[LogBeanV2])
} catch {
case e: JSONException => {
logger.error("parse json exception, error line is : " + line)
}
}
logBean
})
//过滤订单相关的数据,支付成功的数据
val filtered: RDD[LogBeanV2] = logBeanRDD.filter(bean => bean != null && bean.pay_status == 1)
5.2 用户成交金额
5.3 计算各个省的成交金额
5.4 计算各个省下市成交金额的TopN
5.5 计算各个分类成交的TopN
5.6 计算复购率（比较难）
以上所有解如下
LogBeanV2(用于封装日志中的各个属性)
case class LogBeanV2(
page: String,
event_type: Int,
pay_status: Int,
oid: String,
goods: Array[Good],
total_money: Double,
longitude: Double,
latitude:Double,
province: String,
city: String
)
Good（用于封装LogBeanV2中goods属性中的值）
case class Good(
money: Double,
pid: String,
cid: String,
title: String
)
CalculateUtils（所有计算都被封装到次类中）
package cn._51doit.spark.day06
import java.sql.{Date, DriverManager}
import org.apache.spark.rdd.RDD
object CalculateUtils {
//计算复购率
//在一段时间之内，购买两次即以上的用户
def calculateReBuyRatio(filtered: RDD[LogBeanV2]) = {
val uidAndPayCounts = filtered.map(bean => {
val openid = bean.openid
val date = bean.time.split(" ")(0)
((date, openid), 1)
}).reduceByKey(_ + _)
//关联规则
val levelAndCounts = uidAndPayCounts.map(t => {
//根据用户的购买次数计算会员等级的规则
val level = LevelUtil.getLevel(t._2)
val date = t._1._1
((date, level), 1)
}).reduceByKey(_ + _)
//将日期当做Key
//[(2019-09-25, (L1, 5)), (2019-09-25, (L2-3, 3)), (2019-09-25, (L4-5, 4))]
val dateLevelAndCounts = levelAndCounts.map(t => (t._1._1, (t._1._2, t._2)))
//计算出某一天总的购买用户数量
//[{2019-09-25，12}, {2019-09-26，13}]
val datePayUser = uidAndPayCounts.map(t => {
(t._1._1, 1)
}).reduceByKey(_ + _)
//{2019-09-25 -> ((L2-3, 3), 12)}
val joined: RDD[(String, ((String, Int), Int))] = dateLevelAndCounts.join(datePayUser)
val results: RDD[(String, Iterable[(String, String, Double)])] = joined.map(t => {
val date = t._1
val level = t._2._1._1
val levelCouts = t._2._1._2
val totalUser = t._2._2
(date, level, levelCouts.toDouble / totalUser)
}).groupBy(_._1)
//{(2019-09-25, "L1", 0.6), (2019-09-25, "L4-5", 0.3) , (2019-09-25, "L2-3", 0.1)}
val resArr = results.collect()
//获取一个数据库连接
val connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/bigdata?characterEncoding=UTF-8", "root", "123456")
val preparedStatement = connection.prepareStatement("INSERT INTO daily_repay_ratio (`dt`, `L1`, `L2-3`, `L4-5`, `L6+`) VALUES (?, ?, ?, ?, ?)")
resArr.foreach(t => {
val date = t._1
preparedStatement.setString(1, date)
//{(2019-09-25, "L1", 0.6), (2019-09-25, "L2-3", 0.1), (2019-09-25, "L4-5", 0.3)
val list = t._2.toList.sortBy(t => t._2)
for(e <- 0 until list.size) {
//val level = e._2
var tp: (String, String, Double) = (null, null, 0.0)
try {
tp = list(e)
} catch {
case e: Exception => {
}
}
preparedStatement.setDouble(e + 2, tp._3)
}
preparedStatement.executeUpdate()
})
preparedStatement.close()
connection.close()
}
def calculateProvicneAndCityIncomeTopN(filtered: RDD[LogBeanV2]) = {
val reduced: RDD[((String, String), Double)] = filtered.map(bean => {
val province = bean.province
val city = bean.city
val total_money = bean.total_money
((province, city), total_money)
}).reduceByKey(_ + _)
//按照省份进行分组
val result = reduced.groupBy(_._1._1).mapValues(it => it.toList.sortBy(-_._2).take(3))
//将结果写入到数据库
val r = result.collect()
val connection = DriverManager.getConnection("", "", "")
//将数据写入
connection.close()
}
def calculateCategoryIncome(filtered: RDD[LogBeanV2], categoryRDD: RDD[(Int, String)]) = {
val cidAndMoney = filtered.flatMap(bean => {
val goods = bean.goods
goods.map(g => {
val cid = g.cid
val money = g.money
(cid.toInt, money)
})
}).reduceByKey(_ + _)
//
val joined: RDD[(Int, (Double, String))] = cidAndMoney.join(categoryRDD)
val cnameAndMoney: RDD[(String, Double)] = joined.map(t => (t._2._2, t._2._1))
//将数据写入到数据库
cnameAndMoney.foreachPartition(it => {
//获取一个链接
val connection = DriverManager.getConnection("", "", "")
it.foreach(t => {
})
connection.close()
})
}
//计算省份成交金额
def calculateProvinceIncome(filtered: RDD[LogBeanV2]) = {
val provinceAndMoney = filtered.map(bean => {
val province = bean.province
val total_money = bean.total_money
(province, total_money)
}).reduceByKey(_ + _)
//假设数据量比较大，收集到Driver端后再写入，对Driver压力比较大并且写入的效率低
provinceAndMoney.foreachPartition(it => {
val connection = DriverManager.getConnection("", "", "")
val preparedStatement = connection.prepareStatement("INSERT INTO t_province_daily_income VALUES (?, ?, ?)")
it.foreach(t => {
preparedStatement.setDate(1, new Date(System.currentTimeMillis()))
preparedStatement.setString(2, t._1)
preparedStatement.setDouble(3, t._2)
preparedStatement.executeUpdate()
})
preparedStatement.close()
connection.close()
})
}
//计算总的成交金额
def calculateTotalIncome(filtered: RDD[LogBeanV2]) = {
//在Dirver端写入到MySQL中
//sum是一个Action，将计算好的结果收集回Driver
val totalIncome = filtered.map(_.total_money).sum()
//2019-09-26 1000000
//2019-09-27 1100000
val connection = DriverManager.getConnection("", "", "")
val preparedStatement = connection.prepareStatement("INSERT INTO t_daily_income VALUES (?, ?)")
preparedStatement.setDate(1, new Date(System.currentTimeMillis()))
preparedStatement.setDouble(2, totalIncome)
preparedStatement.execute()
preparedStatement.close()
connection.close()
}
}
IncomeCountAdv（计算逻辑）
object IncomeCountAdv {
private val logger: Logger = LoggerFactory.getLogger(this.getClass)
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setAppName(this.getClass.getSimpleName).setMaster("local[*]")
val sc = new SparkContext(conf)
//指定以后从哪里读取数据
val lines: RDD[String] = sc.textFile(args(0))
//整理数据，解析JSON
val logBeanRDD: RDD[LogBeanV2] = lines.map(line => {
var logBean: LogBeanV2 = null
try {
logBean = JSON.parseObject(line, classOf[LogBeanV2])
} catch {
case e: JSONException => {
logger.error("parse json exception, error line is : " + line)
}
}
logBean
})
//过滤订单相关的数据,支付成功的数据
val filtered: RDD[LogBeanV2] = logBeanRDD.filter(bean => bean != null && bean.pay_status == 1)
filtered.cache()
//计算总的成交金额
CalculateUtils.calculateTotalIncome(filtered)
//计算各个省份的成交金额
CalculateUtils.calculateProvinceIncome(filtered)
//计算各个分类的成交金额
val categoryRDD: RDD[(Int, String)] = sc.parallelize(List((1,"图书"), (3,"家具"), (2, "服装"), (4, "手机")))
CalculateUtils.calculateCategoryIncome(filtered, categoryRDD)
//更各个省份下市成交金额的TopN
CalculateUtils.calculateProvicneAndCityIncomeTopN(filtered)
//计算复购率
CalculateUtils.calculateReBuyRatio(filtered)
sc.stop()
}
}