函数传递（序列化）

初始化工作是在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类型，能够序列化。

函数传递（序列化）

推荐阅读更多精彩内容