自定义分组求取topN
程序员文章站
2022-04-21 11:55:49
...
分组是mapreduce当中reduce端的一个功能组件,主要的作用是决定哪些数据作为一组,调用
一次reduce的逻辑,默认是每个不同的key,作为多个不同的组,每个组调用一次reduce逻
辑,我们可以自定义分组实现不同的key作为同一个组,调用一次reduce逻辑
3.2 分析
1、利用“订单id和成交金额”作为key,可以将map阶段读取到的所有订单数据按照id分区,按
照金额排序,发送到reduce
2、在reduce端利用分组将订单id相同的kv聚合成组,然后取第一个即是最大值
第一步:定义OrderBean
定义一个OrderBean,里面定义两个字段,第一个字段是我们的orderId,第二个字段是我们的
金额(注意金额一定要使用Double或者DoubleWritable类型,否则没法按照金额顺序排序
package com.hadoop.demo;
import org.apache.hadoop.io.WritableComparable;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
public class OrderBean implements WritableComparable<OrderBean> {
private String orderId;
private Double price;
public String getOrderId() {
return orderId;
}
public void setOrderId(String orderId) {
this.orderId = orderId;
}
public Double getPrice() {
return price;
}
public void setPrice(Double price) {
this.price = price;
}
@Override
public String toString() {
return orderId + "\t" + price ;
}
// 指定排序规则
@Override
public int compareTo(OrderBean o) {
// 先比较订单ID ,如果订单ID一致,则排序订单金额(降序)
int i = this.orderId.compareTo(o.orderId);
if(i == 0){
i = this.price.compareTo(o.price) * -1 ;
}
return i;
}
// 实现对象序列化
@Override
public void write(DataOutput dataOutput) throws IOException {
dataOutput.writeUTF(orderId);
dataOutput.writeDouble(price);
}
// 实现对象反序列化
@Override
public void readFields(DataInput dataInput) throws IOException {
this.orderId = dataInput.readUTF();
this.price = dataInput.readDouble();
}
}
第二步: 定义Mapper类
package com.hadoop.demo;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import java.io.IOException;
public class GroupMapper extends Mapper<LongWritable, Text,OrderBean,Text> {
@Override
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
// 1. 拆分行文本数据,得到订单的ID,订单金额
String[] split = value.toString().split("\t");
//2. 封装OrderBean 得到 K2
OrderBean orderBean = new OrderBean();
orderBean.setOrderId(split[0]);
orderBean.setPrice(Double.valueOf(split[2]));
//3.将K2 和V2 写入上下文中
context.write(orderBean, value);
}
}
第三步:自定义分区
自定义分区,按照订单id进行分区,把所有订单id相同的数据,都发送到同一个reduce中去
package com.hadoop.demo;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;
public class OrderPartition extends Partitioner<OrderBean, Text> {
/**
*
* 分区规则,根据订单的ID实现分区
**/
@Override
public int getPartition(OrderBean orderBean, Text text, int i) {
return (orderBean.getOrderId().hashCode() & 2147483647) % i;
}
}
第四步:自定义分组
按照我们自己的逻辑进行分组,通过比较相同的订单id,将相同的订单id放到一个组里面去,
进过分组之后当中的数据,已经全部是排好序的数据,我们只需要取前topN即可
package com.hadoop.demo;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
//1.继承WriteableComparator
public class OrderGroupComparator extends WritableComparator {
// 2.调用父类的有参构造
public OrderGroupComparator() {
super(OrderBean.class,true);
}
// 3. 指定分组的规则 重写方法
@Override
public int compare(WritableComparable a, WritableComparable b) {
// 3.1 对形参做强制类型转换
OrderBean first = (OrderBean) a;
OrderBean second = (OrderBean) b;
//3.2 指定分组规则
return first.getOrderId().compareTo(second.getOrderId());
}
}
第五步:定义Reducer类
package com.hadoop.demo;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import java.io.IOException;
public class GroupReducer extends Reducer<OrderBean, Text,Text, NullWritable> {
@Override
protected void reduce(OrderBean key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
int i = 0;
// 获取集合中的前N 条数据
for (Text value : values) {
context.write(value,NullWritable.get());
i++;
if(i >= 1){
break;
}
}
}
}
第六步:程序main函数入口
package com.hadoop.demo;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
public class JobMain extends Configured implements Tool {
@Override
public int run(String[] strings) throws Exception {
// 1. 获取Job对象
Job job = Job.getInstance(super.getConf(), "mygroup_job");
// 2. 设置Job任务
// 第一步:设置输入类和输出路径
job.setInputFormatClass(TextInputFormat.class);
TextInputFormat.addInputPath(job,new Path("file:///E:\\input\\mygroup_in"));
// 第二步: 设置Mapper 类和数据类型
job.setMapperClass(GroupMapper.class);
job.setMapOutputKeyClass(OrderBean.class);
job.setMapOutputValueClass(Text.class);
// 第三 四 五 六
// 设置分区
job.setPartitionerClass(OrderPartition.class);
// 设置分组
job.setGroupingComparatorClass(OrderGroupComparator.class);
// 第七步: 设置 Reducer类和数据类型
job.setReducerClass(GroupReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(NullWritable.class);
// 第八步: 设置输出类 和输出的路径
job.setOutputFormatClass(TextOutputFormat.class);
TextOutputFormat.setOutputPath(job,new Path("file:///E:\\out\\mygroup_out"));
// 3. 等待 job任务结束
boolean b = job.waitForCompletion(true);
return b ? 0 : 1;
}
public static void main(String[] args) throws Exception {
Configuration configuration = new Configuration();
int run = ToolRunner.run(configuration, new JobMain(), args);
System.exit(run);
}
}