Spark自定义对象排序及自定义序列化

一、目标对象添加隐式转换函数的方式实现自定义排序

object MySort{//为student类添加隐式转换函数implicit val stuOrdering = new Ordering[student]{override def compare(x: student, y: student) = {//自定义排序格式为：默认按年龄升序排序，当年龄相同时，按分数降序排序if(x.age != y.age){x.age - y.age}else{y.score - x.score}}}
}/*** 隐式转换函数实现自定义排序*/
object CustomSort1{def main(args: Array[String]): Unit = {//1.创建配置对象val conf = new SparkConf().setAppName("sort1").setMaster("local[2]");//2.创建集群入口类对象val sc = new SparkContext(conf);//3.并行化的方式生成rddval rdd = sc.parallelize(Array(("tom",24,92),("ko",24,97),("mark",28,88),("jack",18,86)))//4.对数据进行排序//引入隐式转换函数import MySort.stuOrderingval resRdd = rdd.sortBy(x => student(x._1,x._2.toInt,x._3.toInt))//5.打印排序后的结果数据println(resRdd.collect().toBuffer)}
}
//学生类
case class student(name:String,age:Int,score:Int)

二、目标对象实现Ordered特质的方式实现自定义排序

object CustomSort2{def main(args: Array[String]): Unit = {//1.创建配置对象val conf = new SparkConf().setAppName("sort2").setMaster("local[2]");//2.创建集群入口类对象val sc = new SparkContext(conf);//3.并行化的方式生成rddval rdd = sc.parallelize(Array(("tom",24,92),("ko",24,97),("mark",28,88),("jack",18,86)))//4.对数据进行排序val resRdd = rdd.sortBy(x => student2(x._1,x._2.toInt,x._3.toInt))//5.打印排序后的结果数据println(resRdd.collect().toBuffer)}
}case class student2(name:String,age:Int,score:Int) extends Ordered[student2]{override def compare(that: student2): Int = {if(this.age != that.age){this.age - that.age}else{that.score - this.score}}
}

三、其他序列化

package com.fuge.bigdata.datahub.analysisimport java.io.{DataInput, DataOutput}import com.fuge.bigdata.tools.common.utils.SparkUtils
import org.apache.hadoop.io.{NullWritable, WritableComparable}
import org.apache.spark.SparkContext/*** Created by chen xiang on 18-6-13.* 一个使用SequenceFile进行存储读取的使用示例*/
object SequenceFileUsage {def main(args: Array[String]): Unit = {require(args.length == 1)// 构建SparkContext对象，封装过，单独运行，自行修改后定义val sc = new SparkContext(SparkUtils.getSparkConf("SequenceFileUsage"))// 获取路径参数val path = args(0).trim// 定义测试数据val studentList = List(Student("01", "abc"), Student("02", "baby"), Student("03", "xiang"))// 序列化测试数据到RDD,并写入到bossc.parallelize(studentList).repartition(1)// 以NullWritable 为key,构建kv结构.SequenceFile需要kv结构才能存储,NullWritable不占存储.map(NullWritable.get() -> _)
// 压缩参数可选用.saveAsSequenceFile(s"$path", Option(classOf[GzipCodec]))// 读取刚才写入的数据val studentRdd = sc.sequenceFile(s"$path/part-*", classOf[NullWritable], classOf[Student]).map {// 读取数据,并且重新赋值对象case (_, y) => Student(y.id, y.name)}.persist()studentRdd.foreach(x => println("count: " + x.id + "\t" + x.name))}
}case class Student(var id: String, var name: String) extends WritableComparable[Student] {/*** 重写无参构造函数,用于反序列化时的反射操作*/def this() {this("", "")}/*** 继承Comparable接口需要实现的方法,用于比较两个对象的大小*/override def compareTo(o: Student): Int = {var cmp = id compareTo o.idif (cmp == 0) {cmp = name compareTo o.name}cmp}/*** 继承Writable接口需要实现的方法-反序列化读取结果,并且赋值到对象字段* 注意要和write的顺序一致*/override def readFields(in: DataInput): Unit = {name = in.readUTF()id = in.readUTF()println("count: " + "\t id = " + id + "\t name = " + name)}/*** 继承Writable接口需要实现的方法-序列化写操作,将对象字段值写入序列化* 注意要和readFields的顺序一致*/override def write(out: DataOutput): Unit = {out.writeUTF(id)out.writeUTF(name)}
}

补充：

1. 自定义的类需要进行序列化，必须都要实现Writable（或者直接实现Seriable，就不用重写了），一般情况下采用实现WritableComparable的方式，并且实现comparaTo，readFields, write方法，并且提供一个无参构造函数

2. readFields和write方法，里面字段的顺序要保持一致

3. 遇到集合类型，序列化时需要先将集合长度写进去，然后再挨个写集合数据

4. 遇到集合类型，反序列化时需要先读取集合的长度，然后接收数据，如果集合数据类型是自定义类型，还需要先实例化一个无参构造，然后赋值。

5. SequenceFile需要使用KV结构才能调用存储，可以使用一个NullWritable来占位，上诉例子中的K值就是使用的NullWritable进行的

6. sequenceFile序列化后占用的存储空间比较大，有需要的话，可以在存储的时候加上压缩算法，具体使用方式可以见上诉的例子

Spark自定义对象排序及自定义序列化相关推荐

java 自定义对象排序,使用自定义排序顺序对对象的ArrayList进行排序
问题我希望为我的地址簿应用程序实现一个排序功能. 我想排序anArrayList contactArray.Contact是一个包含四个字段的类:名称,家庭电话号码,手机号码和地址.我想排序name ...
.NET/C#中对自定义对象集合进行自定义排序的方法
一个集合可否排序,要看系统知不知道排序的规则,像内建的系统类型,int ,string,short,decimal这些,系统知道怎么排序,而如果一个集合里面放置的是自定义类型,比如自己定义了一个Car ...
ht-8 对arrayList中的自定义对象排序( Collections.sort(ListT list, Comparator? super T c))...
1 package com.iotek.set; 2 3 import java.util.ArrayList; 4 import java.util.Collections; 5 import ja ...
ElasticSearch painless脚本实现自定义打分排序
背景: 遇到了这样一个需求,对于历史购买的商品优先排序自定义打分排序规则自定义打分规则,其实就是可以根据打分规则进行排序.注意:只能对索引的keyword属性进行排序,TEXT类型的要加.keyW ...
[置顶] 深入浅出Javascript（三）创建自定义对象以及属性、方法
怎么样创建一个对象? 利用Object创建自定义对象 JavaScript能够自定义对象来扩展程序的功能,不仅如此,它还能扩展JavaScript提供的内置对象,新增内置对象的属性或方法例如下面代码 ...
按属性对自定义对象的ArrayList进行排序
我读过有关使用Comparator对ArrayList进行排序的信息,但在所有示例中,人们都使用了compareTo ,根据一些研究,它是String的一种方法. 我想按自定义对象的属性之一对Arra ...
flask jsonify之序列化时的default函数、jsonify序列化自定义对象
目录 1.看源码 2.重写默认的default函数,实现自己的序列化机制 3.把对象转化成字典 3.1 __dict__的方式 3.2.定义keys和__getitem__的方式 4.最终的代码实现 ...
java查询结果自定义显示_JPA自定义对象接收查询结果集操作
最近使用JPA的时候,碰到需要自定义查询结果集的场景,网上搜了一下,都是需要自定义方法写一大串代码实现的,太繁琐了,有那时间还不如用mybaits. 用JPA就是要尽量通过声明接口解决持久层问题,要不 ...
python json dumps 自定义_Python json.dumps 自定义序列化操作
def login_ajax(request): if request.method == "GET": return render(request, 'login_ajax.ht ...

Spark自定义对象排序及自定义序列化

Spark自定义对象排序及自定义序列化相关推荐

最新文章

热门文章