简介
上一章介绍了MR的wordcount的demo,这一节我们来进行扩充,逐步将wordcount丰富,拥有更多的业务基础逻辑。
我们目标
- 是实现自主切分partition,生成自定义的数量和数据特征的partition文件。
- 并且实现序列化数据bean,利用writableComparable接口实现升序降序排列。
首先我们看一眼数据,这是我随便敲的。
我们用partition分成两个文件,分别是奇数和偶数。
并且每个文件降序排序。
得到如下
文件一:
文件二:
目录结构
和上一章相比我们增加了两个文件
dataBean 和partitioner
dataBean
package MRcode;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;
import org.apache.hadoop.io.WritableComparable;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
/**
* @Author: Braylon
* @Date: 2020/1/19 10:18
* @Version: 1.0
*/
@Getter
@Setter
@AllArgsConstructor
@NoArgsConstructor
public class dataBean implements WritableComparable<dataBean> {
private Long data0;
public void set(Long data0) {
this.data0 = data0;
}
@Override
public int compareTo(dataBean o) {
return this.data0 > o.getData0() ? -1 : 1;
}
@Override
public void readFields(DataInput dataInput) throws IOException {
this.data0 = dataInput.readLong();
}
@Override
public void write(DataOutput dataOutput) throws IOException {
dataOutput.writeLong(data0);
}
@Override
public String toString() {
return '\001' + data0.toString() + '\001';
}
}
这里我增加了一个dataBean来标记数据,因为每个数据只有一个数字,所以我就只写了一个data0,在真正的应用中肯定不止一个属性。然后需要注意的是,这个bean实现了WritableComparable的接口。
至于为什么要使用这个实现自定义数据的序列化,我看了一个博主写的很好,我借用一下:
- 在进行mapreduce编程时key键往往用于分组或排序,当我们在进行这些操作时Hadoop内置的key键数据类型不能满足需求时,
或针对用例优化自定义数据类型可能执行的更好。因此可以通过实现org.apache.hadoop.io.WritableComparable接口定义一个自定义的WritableComparable类型,并使其作为mapreduce计算的key类型。
2.自定义Hadoop key类型。
1.Hadoop mapreduce的key类型往往用于进行相互比较, 可以达到进行相互比较来满足排序的目的。
2.Hadoop Writable数据类型实现了WritableComparable接口,并增加了CompareTo()方法。
CompaeTo()方法的返回值有三种类型。负整数、0、正整数分别对应小于、等于、大于被比较对象
————————————————
原文链接:https://blog.csdn.net/xj626852095/article/details/52675874
然后要注意的就是compareTo方法,这个的返回值很关键,决定了排序方式。
partitioner
package MRcode;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Partitioner;
/**
* @Author: Braylon
* @Date: 2020/1/18 12:03
* @Version: 1.0
*/
public class partitioner extends Partitioner<dataBean, IntWritable> {
@Override
public int getPartition(dataBean dataBean, IntWritable intWritable, int i) {
String str = dataBean.getData0().toString();
String firWord = str.substring(str.length() - 1, str.length());
char[] charArray = firWord.toCharArray();
int result = charArray[0];
if (result % 2 == 0) {
return 0;
}else {
return 1;
}
}
}
这里就是根据数据的最后以为的奇偶来返回到不同的parttion文件中,定义了两种不同的返回数据我们需要在driver中设定reducenum,如下所示:
driver
package MRcode;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import java.io.IOException;
public class WorkCountDriver {
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
args = new String[] {"inputPath","outputPath"};
//获取配置信息
Configuration conf = new Configuration();
Job job = Job.getInstance(conf);
job.setJarByClass(WorkCountDriver.class);
job.setMapperClass(WordCountMapper.class);
job.setReducerClass(WordCountReducer.class);
job.setMapOutputKeyClass(dataBean.class);
job.setMapOutputValueClass(IntWritable.class);
//reduce 输出的K,V类型
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
job.setPartitionerClass(partitioner.class);
job.setNumReduceTasks(2);
FileInputFormat.setInputPaths(job,new Path(args[0]));
FileOutputFormat.setOutputPath(job,new Path(args[1]));
//提交job
job.waitForCompletion(true);
}
}
job.setPartitionerClass(partitioner.class);
job.setNumReduceTasks(2);
上面的两行是很重要的,尤其是设置reducetask的个数,由于在partitioner中设置了两种不同的返回类型,所以这里的个数必须要大于等于2,否则会报错。
下面我放上mapper和reducer类的代码,其中比较重要的是他们的输入输出类型,具体的map和reduce的过程与原理大家可以看上一章我写的,这里不再赘述。
mapper
package MRcode;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import java.io.IOException;
public class WordCountMapper extends Mapper<LongWritable, Text, dataBean, IntWritable> {
dataBean k = new dataBean();
IntWritable v = new IntWritable();
@Override
protected void map(LongWritable key,
Text value,
Context context) throws IOException, InterruptedException {
//第一步 转化格式
String line0 = value.toString();
//第二部 切分数据 按空格切分数据
String[] words = line0.split(" ");
//第三步 输出<单词 , 1>
for (String word : words) {
//写法1:context.write(new Text(word),new IntWritable(1));
k.set(Long.parseLong(word));
v.set(1);
context.write(k,v);
}
}
}
reducer
package MRcode;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import java.io.IOException;
public class WordCountReducer extends Reducer<dataBean, IntWritable,Text,IntWritable> {
@Override
protected void reduce(dataBean key,
Iterable<IntWritable> values,
Context context) throws IOException, InterruptedException {
//初始化次数
int count=0;
//汇总次数
for(IntWritable value : values){
count += value.get();
}
//输出总次数
context.write(new Text(key.getData0().toString()), new IntWritable(count));
}
}
最终得到上述的结果。