1. Task线程安全问题

12.1 现象和原理

在一个Executor可以同时运行多个Task，如果多个Task使用同一个共享的单例对象，如果对共享的数据同时进行读写操作，会导致线程不安全的问题，为了避免这个问题，可以加锁，但效率变低了，因为在一个Executor中同一个时间点只能有一个Task使用共享的数据，这样就变成了串行了，效率低！

12.2 案例

定义一个工具类object，格式化日期，因为SimpleDateFormat线程不安全，会出现异常

Scala
val conf = new SparkConf()
.setAppName("WordCount")
.setMaster("local[*]") //本地模式，开多个线程
//1.创建SparkContext
val sc = new SparkContext(conf)

val lines = sc.textFile("data/date.txt")

val timeRDD: RDD[Long] = lines.map(e => {
//将字符串转成long类型时间戳
//使用自定义的object工具类
val time: Long = DateUtilObj.parse(e)
time
})

val res = timeRDD.collect()
println(res.toBuffer)

Scala
object DateUtilObj {

//多个Task使用了一个共享的SimpleDateFormat，SimpleDateFormat是线程不安全

val *sdf *= new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")

//线程安全的
//val sdf: FastDateFormat = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss")

def parse(str: String): Long = {
//2022-05-23 11:39:30
sdf.parse(str).getTime
}

}

上面的程序会出现错误，因为多个Task同时使用一个单例对象格式化日期，报错，如果加锁，程序会变慢，改进后的代码：

Scala
val conf = new SparkConf()
.setAppName("WordCount")
.setMaster("local[*]") //本地模式，开多个线程
//1.创建SparkContext
val sc = new SparkContext(conf)

val lines = sc.textFile("data/date.txt")

val timeRDD = lines.mapPartitions(it => {
//一个Task使用自己单独的DateUtilClass实例，缺点是浪费内存资源
val dataUtil = new DateUtilClass
it.map(e => {
dataUtil.parse(e)
})
})

val res = timeRDD.collect()
println(res.toBuffer)

Scala
class DateUtilClass {

val *sdf *= new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")

def parse(str: String): Long = {
//2022-05-23 11:39:30
sdf.parse(str).getTime
}
}

改进后，一个Task使用一个DateUtilClass实例，不会出现线程安全的问题。

1. 累加器

累加器是Spark中用来做计数功能的，在程序运行过程当中，可以做一些额外的数据指标统计

需求：在处理数据的同时，统计一下指标数据，具体的需求为：将RDD中对应的每个元素乘以10，同时在统计每个分区中偶数的数据

13.1 不使用累加器的方案

需要多次触发Action，效率低，数据会被重复计算

Scala
/**

• • 不使用累加器，而是触发两次Action*

• /
  object C12_AccumulatorDemo1 {

  def main(args: Array[String]): Unit = {

  val conf = new SparkConf()
  .setAppName("WordCount")
  .setMaster("local[*]") //本地模式，开多个线程
  //1.创建SparkContext
  val sc = new SparkContext(conf)

  val rdd1 = sc.parallelize(List(1,2,3,4,5,6,7,8,9), 2)
  //对数据进行转换操作（将每个元素乘以10），同时还要统计每个分区的偶数的数量
  val rdd2 = rdd1.map(_ * 10)
  //第一次触发Action
  rdd2.saveAsTextFile("out/111")

  //附加的指标统计
  val rdd3 = rdd1.filter(_ % 2 == 0)
  //第二个触发Action
  val c = rdd3.count()
  println(c)
  }

}

13.2 使用累加器的方法

触发一次Action，并且将附带的统计指标计算出来，可以使用Accumulator进行处理，Accumulator的本质数一个实现序列化接口class，每个Task都有自己的累加器，避免累加的数据发送冲突

Scala
object C14_AccumulatorDemo3 {

def main(args: Array[String]): Unit = {

 val conf = new SparkConf()
   .setAppName("WordCount")
   .setMaster("local[*]") //本地模式，开多个线程
 //1.创建SparkContext
 val sc = new SparkContext(conf)

 val rdd1 = sc.parallelize(*List*(1,2,3,4,5,6,7,8,9), 2)
 //在Driver定义一个特殊的变量，即累加器
 //Accumulator可以将每个分区的计数结果，通过网络传输到Driver，然后进行全局求和
 val accumulator: LongAccumulator = sc.longAccumulator("even-acc")
 val rdd2 = rdd1.map(e => {
   if (e % 2 == 0) {
     accumulator.add(1)  //闭包，在Executor中累计的
   }
   e * 10
 })

 //就触发一次Action
 rdd2.saveAsTextFile("out/113")

 //每个Task中累计的数据会返回到Driver吗？
 *println*(accumulator.count)

}
}

1. StandAlone的两种执行模式

spark自动的StandAlone集群有两种运行方式，分别是client模式和cluster模式，默认使用的是client模式。两种运行模式的本质区别是，Driver运行在哪里了

14.1 什么是Driver

Driver本意是驱动的意思（类似叫法的有MySQL的连接驱动），在就是与集群中的服务建立连接，执行一些命令和请求的。但是在Spark的Driver指定就是SparkContext和里面创建的一些对象，所有可以总结为，SparkContext在哪里创建，Driver就在哪里。Driver中包含很多的对象实例，有SparkContext，DAGScheduler、TaskScheduler、ShuffleManager、BroadCastManager等，Driver是对这些对象的统称。

14.2 client模式

Driver运行在用来提交任务的SparkSubmit进程中，在Spark的stand alone集群中，提交spark任务时，可以使用cluster模式即--deploy-mode client （默认的）

注意：spark-shell只能以client模式运行，不能以cluster模式运行，因为提交任务的命令行客户端和SparkContext必须在同一个进程中。

14.3 cluster模式

Driver运行在Worker启动的一个进程中，这个进程叫DriverWapper，在Spark的stand alone集群中，提交spark任务时，可以使用cluster模式即--deploy-mode cluster

特点：Driver运行在集群中，不在SparkSubmit进程中，需要将jar包上传到hdfs中

Shell
spark-submit --master spark://node-1.51doit.cn:7077 --class cn._51doit.spark.day01.WordCount --deploy-mode cluster hdfs://node-1.51doit.cn:9000/jars/spark10-1.0-SNAPSHOT.jar hdfs://node-1.51doit.cn:9000/wc hdfs://node-1.51doit.cn:9000/out002

cluster模式的特点：可以给Driver更灵活的指定一些参数，可以给Driver指定内存大小，cores的数量

如果一些运算要在Driver进行计算，或者将数据收集到Driver端，这样就必须指定Driver的内存和cores更大一些

Shell

指定Driver的内存，默认是1g

--driver-memory MEM Memory for driver (e.g. 1000M, 2G) (Default: 1024M).

指定Driver的cores，默认是1

--driver-cores NUM Number of cores used by the driver, only in cluster mode (Default: 1).