初始化工作是在Driver端进行的，而实际运行程序是在Excutor端进行的。
跨进程通信，是需要序列化的
算子前面的在Driver执行
算子里面的在Excutor执行
spark在执行作业之前，会先进行闭包检测，目的在于对闭包所使用的变量是否序列化进行检测。
package com.atguigu
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object Action {
def main(args: Array[String]): Unit = {
val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("ADCount")
val sc = new SparkContext(conf)
val rdd: RDD[String] = sc.parallelize(Array("hadoop","spark","hive","atguigu"))
//创建一个Search对象
val search = new Search("h")
//使用第一个过滤函数并打印结果
val match1RDD: RDD[String] = search.getMatch1(rdd)
match1RDD.collect().foreach(println)
}
}
/***************************************/
class Search(query:String) {
//过滤出包含字符串的数据
def isMatch(s:String): Boolean ={
s.contains(query)
}
//过滤出包含字符串的RDD
def getMatch1(rdd: RDD[String]):RDD[String] = {
rdd.filter(isMatch)
}
//过滤出包含字符串的RDD
def getMatch2(rdd:RDD[String]):RDD[String] = {
rdd.filter(x => x.contains(query))//query在Driver端创建的, 如果使用getMatch2没有序列化也会报错。
}
}
报错：
Exception in thread "main" org.apache.spark.SparkException: Task not serializable
-object not serializable (class: com.atguigu.Search, value: com.atguigu.Search@660591fb)
Search对象未序列化错误。原因是： val search = new Search("h")在Driver端声明，所以Seach对象在Driver端，而getMatch1中的算子filter中的（isMatch)在算子里面，所以在Excutor端执行。需要将在Driver端的Seach对象发送到Excutor端，所以需要序列化。
混入Serializable特质，实现序列化。
class Search(query:String) extends Serializable {
//过滤出包含字符串的数据
def isMatch(s:String): Boolean ={
s.contains(query)
}
//过滤出包含字符串的RDD
def getMatch1(rdd: RDD[String]):RDD[String] = {
rdd.filter(isMatch)
}
//过滤出包含字符串的RDD
def getMatch2(rdd:RDD[String]):RDD[String] = {
rdd.filter(x => x.contains(query))
}
}
hadoop
hive
二、
package com.atguigu
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object Action {
def main(args: Array[String]): Unit = {
val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("ADCount")
val sc = new SparkContext(conf)
val rdd: RDD[String] = sc.parallelize(Array("hadoop","spark","hive","atguigu"))
//创建一个Search对象
val search = new Search("h")
//使用第一个过滤函数并打印结果
val match2RDD: RDD[String] = search.getMatch2(rdd)
match2RDD.collect().foreach(println)
}
}
class Search(query:String) {
//过滤出包含字符串的数据
def isMatch(s:String): Boolean ={
s.contains(query)
}
//过滤出包含字符串的RDD
def getMatch1(rdd: RDD[String]):RDD[String] = {
rdd.filter(isMatch)
}
//过滤出包含字符串的RDD
def getMatch2(rdd:RDD[String]):RDD[String] = {
val q = query
rdd.filter(x => x.contains(q))
}
}
hadoop
hive
为什么 val q = query中间变量传递以下就没有出错了呢？
原因：x => x.contains(q)在Excutor端执行，q在算子filter之外声明的，也就是在Driver端声明的，将q传往Excutor端，为啥不要序列化呢？因为q是一个String类型，能够序列化。
