Task序列化问题
问题原因
一个Executor是一个进程，一个进程中可以同时运行多个Task，如果多个Task使用了共享的变量，就会出现线程不安全的问题
案例
需求
使用spark将日期字符串转换成long类型时间戳
样例数据
2019-11-06 15:59:50
2019-11-06 15:59:51
2019-11-06 15:59:52
2019-11-06 15:59:53
2019-11-06 15:59:54
2019-11-06 15:59:55
2019-11-06 15:59:56
2019-11-06 15:59:57
2019-11-06 15:59:58
代码实现一
多个Task使用一个单例的工具类，工具类使用共享的SimpleDateFormat，会出现线程不安全的问题，所有方法要加锁，但是效率变低
object DateUtils {val sdf1 = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")//为了避免多个线程同时使用一个SimpleDateFormat出现问题，加锁def parse1(line: String): Long = synchronized {val date = sdf1.parse(line)date.getTime}//FastDateFormat内部实现了锁机制val sdf2 = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss")def parse2(line: String): Long = {val date = sdf2.parse(line)date.getTime}
}
object TaskThreadNotSafe01 {def main(args: Array[String]): Unit = {val isLocal = args(0).toBoolean//创建SparkConf，然后创建SparkContextval conf = new SparkConf().setAppName(this.getClass.getSimpleName)if (isLocal) {conf.setMaster("local[*]")}val sc = new SparkContext(conf)val lines = sc.textFile(args(1))val timeRdd: RDD[Long] = lines.map(line => {val time: Long = DateUtils.parse2(line)time})//触发Actionval r = timeRdd.collect()println(r.toBuffer)sc.stop()}
}
代码实现二
- 在driver端new一个类的实例，然后在函数中使用，一个Task有一个工具类的实例
- 工具类要实现序列化
class DateUtilsClass extends Serializable {val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")def parse(line: String): Long = {val date = sdf.parse(line)date.getTime}
}object TaskThreadNotSafe02 {def main(args: Array[String]): Unit = {val isLocal = args(0).toBoolean//创建SparkConf，然后创建SparkContextval conf = new SparkConf().setAppName(this.getClass.getSimpleName)if (isLocal) {conf.setMaster("local[*]")}val sc = new SparkContext(conf)//在Driver端被创建的【new一个类的实例】val dateUtils = new DateUtilsClassval lines = sc.textFile(args(1))val timeRdd: RDD[Long] = lines.map(line => {val time: Long = dateUtils.parse(line)time})//触发Actionval r = timeRdd.collect()println(r.toBuffer)sc.stop()}
}
代码实现三
- 使用MapPartitions方法，工具类不用实现序列化
- 每一个Task都有一个但是的工具类实例
class DateUtilsClassNoSer {val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")def parse(line: String): Long = {val date = sdf.parse(line)date.getTime}
}object TaskThreadNotSafe03 {def main(args: Array[String]): Unit = {val isLocal = args(0).toBoolean//创建SparkConf，然后创建SparkContextval conf = new SparkConf().setAppName(this.getClass.getSimpleName)if (isLocal) {conf.setMaster("local[*]")}val sc = new SparkContext(conf)val lines = sc.textFile(args(1))val timeRdd = lines.mapPartitions(it => {//一个分区new一个DateUtilsClassNoSer工具类val dateUtils = new DateUtilsClassNoSerit.map(line => {dateUtils.parse(line)})})//触发Actionval r = timeRdd.collect()println(r.toBuffer)sc.stop()}
}