Spark-Core RDD中函数（变量）传递
2024-09-05 19:27:08
我们进行 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都可以使用样例类
最新文章
- Hyperledger区块数据的访问
- Jconsole远程监控tomcat 的JVM内存(linux、windows)
- MAC Java 开发环境配置
- 使用PHP编写发红包程序
- 读《JavaScript语言精粹》的一些感言
- 【WPF】Application应用程序启动
- Ubuntu 12.04搭建Andorid编译环境
- android135 360 来电去电归属地显示，自定义toast,
- careercup-递归和动态规划 9.9
- Visual Studio 单元测试之五---数据库测试
- sql还原(.mdf文件还原)
- mysql时间戳与日期格式的相互转换
- win7系统中使用DOS命令是出现乱码的解决方法
- python 中的 yield 究竟为何物？生成器和迭代器的区别？
- HTMLParser和BeautifulSoup使用入门和总结
- 从github clone文件： Failed to receive SOCKS4 connect request ack.
- 利用RecyclerView实现无限轮播广告条
- python标准库介绍——36 popen2 模块详解
- 【delphi】TStringList类常用属性方法详解
- ffmpeg 将jpg转为yuv