Spark-Core RDD中函数（变量）传递

矫情吗；* 2023-06-23 03:59 3阅读 0赞

我们进行 Spark 进行编程的时候，**初始化工作是在driver端完成的**，而**实际的运行程序是在executor端进行**，所以就涉及到了进程间的通讯，数据是需要序列化的

### 1、传递函数 ###

import org.apache.spark.{SparkConf, SparkContext}
    import org.apache.spark.rdd.RDD
    
    object SerDemo {
        def main(args: Array[String]): Unit = {
            val conf: SparkConf = new SparkConf().setAppName("SerDemo").setMaster("local[*]")
            val sc = new SparkContext(conf)
            val rdd: RDD[String] = sc.parallelize(Array("hello world", "hello h", "h", "hahah"), 2)
            val searcher = new Searcher("hello")
            val result: RDD[String] = searcher.getMatchedRDD1(rdd)
            result.collect.foreach(println)
        }
    }
    //需求: 在 RDD 中查找出来包含 query 子字符串的元素
    
    // query 为需要查找的子字符串
    class Searcher(val query: String){
        // 判断 s 中是否包括子字符串 query
        def isMatch(s : String) ={
            s.contains(query)
        }
        // 过滤出包含 query字符串的字符串组成的新的 RDD
        def getMatchedRDD1(rdd: RDD[String]) ={
            rdd.filter(isMatch)  //
        }
    }

**说明：**

（1）直接运行程序会报错：`没有初始化`。因为`rdd.filter(isMatch)`用到了`对象this的方法isMatch`，所以对象**this需要序列化**，**才能把对象从driver发送到executor**

（2）解决方法：让 Searcher 类实现序列化接口:`Serializable`

// query 为需要查找的子字符串
    class Searcher(val query: String) extends Serializable{
        // 判断 s 中是否包括子字符串 query
        def isMatch(s : String) ={
            s.contains(query)
        }
        // 过滤出包含 query字符串的字符串组成的新的 RDD
        def getMatchedRDD1(rdd: RDD[String]) ={
            rdd.filter(isMatch)  //
        }
    }

### 2、传递变量 ###

// query 为需要查找的子字符串
    class Searcher(val query: String){
        // 过滤出包含 query字符串的字符串组成的新的 RDD
        def getMatchedRDD2(rdd: RDD[String]) ={
            rdd.filter(_.contains(query))
        }
    }

**说明：**

（1）这次没有传递函数, 而是传递了一个属性过去. 仍然会报错没有序列化. **因为this仍然没有序列化**（`属性query也是属于this`）

（2）解决方法1：让类实现序列化接口:Serializable

（3）解决方法2：传递局部变量而不是属性

// query 为需要查找的子字符串
    class Searcher(val query: String){
        // 过滤出包含 query字符串的字符串组成的新的 RDD
        def getMatchedRDD2(rdd: RDD[String]) ={
            var q = query
            rdd.filter(_.contains(q))
        }
    }

### 3、kryo序列化框架 ###

Java 的序列化比较重, 能够序列化任何的类. 比较灵活,但是相当的慢, 并且序列化后对象的提交也比较大.

Spark 处于性能的考虑, 支持另外一种序列化机制: kryo (2.0开始支持). kryo 比较快和简洁.(速度是Serializable的10倍). 想获取更好的性能应该使用 kryo 来序列化.

从`2.0`开始, **Spark 内部已经在使用 kryo 序列化机制**: 当 RDD 在`Shuffle数据`的时候, **简单数据类型, 简单数据类型的数组和字符串类型已经在使用 kryo 来序列化.**

有一点需要注意的是: **即使使用 kryo 序列化, 也要继承 Serializable 接口.**

import org.apache.spark.{SparkConf, SparkContext}
    import org.apache.spark.rdd.RDD
    
    object SerDemo {
        def main(args: Array[String]): Unit = {
            val conf: SparkConf = new SparkConf()
                .setAppName("SerDemo")
                .setMaster("local[*]")
                // 替换默认的序列化机制
                .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
                // 注册需要使用 kryo 序列化的自定义类
                .registerKryoClasses(Array(classOf[Searcher]))
            val sc = new SparkContext(conf)
            val rdd: RDD[String] = sc.parallelize(Array("hello world", "hello h", "h", "h"), 2)
            val searcher = new Searcher("hello")
            val result: RDD[String] = searcher.getMatchedRDD1(rdd)
            result.collect.foreach(println)
        }
    }
    
     class Searcher(val query: String) extends Serializable{
        // 判断 s 中是否包括子字符串 query
        def isMatch(s: String) = {
            s.contains(query)
        }
        // 过滤出包含 query字符串的字符串组成的新的 RDD
        def getMatchedRDD1(rdd: RDD[String]) = {
            rdd.filter(isMatch) //
        }
    
        // 过滤出包含 query字符串的字符串组成的新的 RDD
        def getMatchedRDD2(rdd: RDD[String]) = {
            val q = query
            rdd.filter(_.contains(q))
        }
    }

### 4、以上`extends Serializable`都可以使用样例类 ###