备案ip 查询网站查询网站,企业管理软件,东营抖音代运营,公司网站设计报价二次排序就是首先按照第一字段排序#xff0c;然后再对第一字段相同的行按照第二字段排序#xff0c;注意不能破坏第一次排序的结果。 这里主要讲如何使用一个Mapreduce就可以实现二次排序。Hadoop有自带的SecondarySort程序#xff0c;但这个程序只能对整数进行排序#x… 二次排序就是首先按照第一字段排序然后再对第一字段相同的行按照第二字段排序注意不能破坏第一次排序的结果。 这里主要讲如何使用一个Mapreduce就可以实现二次排序。Hadoop有自带的SecondarySort程序但这个程序只能对整数进行排序所以我们需要对其进行改进使其可以对任意字符串进行排序。下面会分别列出这两个程序的详解。 Hadoop自带的例子中定义的map和reduce如下关键是它对输入输出类型的定义java泛型编程 public static class Map extends MapperLongWritable, Text, IntPair, IntWritable public static class Reduce extends ReducerIntPair, NullWritable, IntWritable, IntWritable 在 map阶段使用job.setInputFormatClass定义的InputFormat将输入的数据集分割成小数据块splites同时 InputFormat提供一个RecordReder的实现。本例子中使用的是TextInputFormat他提供的RecordReder会将文 本的一行的行号作为key这一行的文本作为value。这就是自定义Map的输入是LongWritable, Text的原因。然后调用自定义Map的map方法将一个个LongWritable, Text对输入给Map的map方法。注意输出应该符合自定义Map中定义的输出IntPair, IntWritable。最终是生成一个ListIntPair, IntWritable。在map阶段的最后会先调用job.setPartitionerClass对这个List进行分区每个分区映射到 一个reducer。每个分区内又调用job.setSortComparatorClass设置的key比较函数类排序。可以看到这本身就是一个二次 排序。如果没有通过job.setSortComparatorClass设置key比较函数类则使用key的实现的compareTo方法。在第一个 例子中使用了IntPair实现的compareTo方法而在下一个例子中专门定义了key比较函数类。 在reduce阶 段reducer接收到所有映射到这个reducer的map输出后也是会调用job.setSortComparatorClass设置的key比 较函数类对所有数据对排序。然后开始构造一个key对应的value迭代器。这时就要用到分组使用 jobjob.setGroupingComparatorClass设置的分组函数类。只要这个比较器比较的两个key相同他们就属于同一个组它们 的value放在一个value迭代器而这个迭代器的key使用属于同一个组的所有key的第一个key。最后就是进入Reducer的reduce方 法reduce方法的输入是所有的key和它的value迭代器。同样注意输入与输出的类型必须与自定义的Reducer中声明的一致。 2 Hadoop自带的只对两个整型自带排序例子详解 2.1 测试数据如下所示 20 2150 5150 5250 5350 5460 5160 5360 5260 5660 5770 5860 6170 5470 5570 5670 5770 581 23 45 67 82203 2150 51250 52250 53530 5440 51120 5320 52260 5660 57740 5863 61730 5471 5571 5673 5774 5812 21131 4250 627 8 import java.io.DataInput;
import java.io.DataOutput;
import java.io.File;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.StringTokenizer;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Job.JobState;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;import service.plugin.EJob;public class SecondarySort{/*** ClassName IntPair* Description 定义IntPair对象该对象实现WritableComparable接口描述第一列和第二列数据同时完成两列数据的相关操作这里是对二者进行比较* */public static class IntPair implements WritableComparableIntPair {int first;int second;/*** Set the left and right values.*/public void set(int left, int right) {first left;second right;}public int getFirst() {return first;}public int getSecond() {return second;}Override// 反序列化从流中的二进制转换成IntPairpublic void readFields(DataInput in) throws IOException {// TODO Auto-generated method stubfirst in.readInt();second in.readInt();}Override// 序列化将IntPair转化成使用流传送的二进制public void write(DataOutput out) throws IOException {// TODO Auto-generated method stubout.writeInt(first);out.writeInt(second);}Override// key的比较public int compareTo(IntPair o) {// TODO Auto-generated method stubif (first ! o.first) {return first o.first ? -1 : 1;} else if (second ! o.second) {return second o.second ? -1 : 1;} else {return 0;}}// 新定义类应该重写的两个方法不用这个方法好像也可以// Override// The hashCode() method is used by the HashPartitioner (the default// partitioner in MapReduce)// public int hashCode() {// return first * 157 second;// }Overridepublic boolean equals(Object right) {if (right null)return false;if (this right)return true;if (right instanceof IntPair) {IntPair r (IntPair) right;return r.first first r.second second;} else {return false;}}}/*** 分区函数类。根据first确定Partition。*/public static class FirstPartitioner extends PartitionerIntPair, IntWritable {Overridepublic int getPartition(IntPair key, IntWritable value, int numPartitions) {System.out.println(FirstPartitioner-----------------------------------------------);System.out.println(Math.abs(key.getFirst() * 127) % numPartitions: Math.abs(key.getFirst() * 127) % numPartitions);return Math.abs(key.getFirst() * 127) % numPartitions;}}/*** 分组函数类。只要first相同就属于同一个组。*//** //第一种方法实现接口RawComparator public static class GroupingComparator* implements RawComparatorIntPair {* * Override public int compare(IntPair o1, IntPair o2) { int l * o1.getFirst(); int r o2.getFirst(); return l r ? 0 : (l r ? -1 :* 1); }* * Override //一个字节一个字节的比直到找到一个不相同的字节然后比这个字节的大小作为两个字节流的大小比较结果。 public int* compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2){ // TODO* Auto-generated method stub return WritableComparator.compareBytes(b1, s1,* Integer.SIZE/8, b2, s2, Integer.SIZE/8); } }*/// 第二种方法继承WritableComparatorpublic static class GroupingComparator extends WritableComparator {protected GroupingComparator() {super(IntPair.class, true);System.out.println(GroupingComparator---------------------------------);}Override// Compare two WritableComparables.public int compare(WritableComparable w1, WritableComparable w2) {IntPair ip1 (IntPair) w1;IntPair ip2 (IntPair) w2;int l ip1.getFirst();int r ip2.getFirst();return l r ? 0 : (l r ? -1 : 1);}}/*** ClassName Map* Description 自定义map类将每行数据进行分拆第一列的数据存入left变量第二列数据存入right变量* 在map阶段的最后会先调用job.setPartitionerClass对这个List进行分区每个分区映射到一个reducer* 。每个分区内又调用job.setSortComparatorClass设置的key比较函数类排序。可以看到这本身就是一个二次排序。*/public static class Map extendsMapperLongWritable, Text, IntPair, IntWritable {private final IntPair intkey new IntPair();private final IntWritable intvalue new IntWritable();public void map(LongWritable key, Text value, Context context)throws IOException, InterruptedException {String line value.toString();// 调用java自己的工具类StringTokenizer(),将map输入的每行字符串按规则进行分割成每个字符串这些规则有\t\n\r\f基本上分割的结果都可以保证到最细的字符串粒度StringTokenizer tokenizer new StringTokenizer(line);int left 0;int right 0;if (tokenizer.hasMoreTokens()) {left Integer.parseInt(tokenizer.nextToken());System.out.println(left: left);if (tokenizer.hasMoreTokens())right Integer.parseInt(tokenizer.nextToken());intkey.set(left, right);intvalue.set(right);context.write(intkey, intvalue);}}}// 自定义reducepublic static class Reduce extendsReducerIntPair, IntWritable, Text, IntWritable {private final Text left new Text();private static final Text SEPARATOR new Text(------------------------------------------------);public void reduce(IntPair key, IterableIntWritable values,Context context) throws IOException, InterruptedException {context.write(SEPARATOR, null);System.out.println(------------------------------------------------);left.set(Integer.toString(key.getFirst()));for (IntWritable val : values) {System.out.println(reduce: left left , val val);context.write(left, val);}}}/*** param args*/public static void main(String[] args) throws IOException,InterruptedException, ClassNotFoundException {// 读取hadoop配置File jarFile EJob.createTempJar(bin);ClassLoader classLoader EJob.getClassLoader();Thread.currentThread().setContextClassLoader(classLoader);Configuration conf new Configuration(true);String[] otherArgs new String[2];otherArgs[0] hdfs://192.168.1.100:9000/test_in/secondary_sort_data.txt;String time new SimpleDateFormat(yyyyMMddHHmmss).format(new Date());otherArgs[1] hdfs://192.168.1.100:9000/test_out/mr- time;Job job new Job(conf, secondarysort);job.setJarByClass(SecondarySort.class);((JobConf) job.getConfiguration()).setJar(jarFile.toString());job.setMapperClass(Map.class);// 不再需要Combiner类型因为Combiner的输出类型Text,// IntWritable对Reduce的输入类型IntPair, IntWritable不适用// job.setCombinerClass(Reduce.class);// 分区函数job.setPartitionerClass(FirstPartitioner.class);// 分组函数job.setGroupingComparatorClass(GroupingComparator.class);// Reducer类型job.setReducerClass(Reduce.class);// map输出Key的类型job.setMapOutputKeyClass(IntPair.class);// map输出Value的类型job.setMapOutputValueClass(IntWritable.class);// reduce输出Key的类型是Text因为使用的OutputFormatClass是TextOutputFormatjob.setOutputKeyClass(Text.class);// reduce输出Value的类型job.setOutputValueClass(IntWritable.class);// 将输入的数据集分割成小数据块splites同时提供一个RecordReder的实现。job.setInputFormatClass(TextInputFormat.class);// 提供一个RecordWriter的实现负责数据输出。job.setOutputFormatClass(TextOutputFormat.class);FileInputFormat.setInputPaths(job, new Path(otherArgs[0]));FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));// 提交jobif (job.waitForCompletion(false)) {System.out.println(job ok !);} else {System.out.println(job error !);}}
} 执行结果如下所示 ------------------------------------------------1 2------------------------------------------------3 4------------------------------------------------5 6------------------------------------------------7 87 82------------------------------------------------12 211------------------------------------------------20 2120 5320 522------------------------------------------------31 42------------------------------------------------40 511------------------------------------------------50 5150 5250 5350 5350 5450 6250 51250 522------------------------------------------------60 5160 5260 5360 5660 5660 5760 5760 61------------------------------------------------63 61------------------------------------------------70 5470 5570 5670 5770 5870 58------------------------------------------------71 5571 56------------------------------------------------73 57------------------------------------------------74 58------------------------------------------------203 21------------------------------------------------530 54------------------------------------------------730 54------------------------------------------------740 58 3 改进后的二次排序可对字符串进行排序 3.1 测试数据如下所示 import javaimport javaimport javaimport java import1 orgimport org1import1 orgimport2 org2import orgimport2 org1import1 orgimport1 orgimport org2import2 org3 orgimport orgimport1 orgimportin orgimport orghello time import java.io.DataInput;
import java.io.DataOutput;
import java.io.File;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;import service.plugin.EJob;public class SecondarySortString {// 自己定义的key类应该实现WritableComparable接口public static class IntPair implements WritableComparableIntPair {String first;String second;/*** Set the left and right values.*/public void set(String left, String right) {first left;second right;}public String getFirst() {return first;}public String getSecond() {return second;}// 反序列化从流中的二进制转换成IntPairpublic void readFields(DataInput in) throws IOException {first in.readUTF();second in.readUTF();}// 序列化将IntPair转化成使用流传送的二进制public void write(DataOutput out) throws IOException {out.writeUTF(first);out.writeUTF(second);}// 重载 compareTo 方法进行组合键 key 的比较该过程是默认行为。// 分组后的二次排序会隐式调用该方法。public int compareTo(IntPair o) {if (!first.equals(o.first)) {return first.compareTo(o.first);} else if (!second.equals(o.second)) {return second.compareTo(o.second);} else {return 0;}}// 新定义类应该重写的两个方法// The hashCode() method is used by the HashPartitioner (the default// partitioner in MapReduce)public int hashCode() {return first.hashCode() * 157 second.hashCode();}public boolean equals(Object right) {if (right null)return false;if (this right)return true;if (right instanceof IntPair) {IntPair r (IntPair) right;return r.first.equals(first) r.second.equals(second);} else {return false;}}}/*** 分区函数类。根据first确定Partition。*/public static class FirstPartitioner extends PartitionerIntPair, Text {public int getPartition(IntPair key, Text value, int numPartitions) {return Math.abs(key.getFirst().hashCode() * 127) % numPartitions;}}/*** 分组函数类。只要first相同就属于同一个组。*//** //第一种方法实现接口RawComparator public static class GroupingComparator* implements RawComparatorIntPair { public int compare(IntPair o1,* IntPair o2) { int l o1.getFirst(); int r o2.getFirst(); return l r* ? 0 : (l r ? -1 : 1); }* //一个字节一个字节的比直到找到一个不相同的字节然后比这个字节的大小作为两个字节流的大小比较结果。 public int* compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2){ return* WritableComparator.compareBytes(b1, s1, Integer.SIZE/8, b2, s2,* Integer.SIZE/8); } }*/// 第二种方法继承WritableComparatorpublic static class GroupingComparator extends WritableComparator {protected GroupingComparator() {super(IntPair.class, true);}// Compare two WritableComparables.// 重载 compare对组合键按第一个自然键排序分组public int compare(WritableComparable w1, WritableComparable w2) {IntPair ip1 (IntPair) w1;IntPair ip2 (IntPair) w2;String l ip1.getFirst();String r ip2.getFirst();return l.compareTo(r);}}// 自定义mappublic static class Map extends MapperLongWritable, Text, IntPair, Text {private final IntPair keyPair new IntPair();String[] lineArr null;public void map(LongWritable key, Text value, Context context)throws IOException, InterruptedException {String line value.toString();if(line.isEmpty()){return;}lineArr line.split( , -1);keyPair.set(lineArr[0], lineArr[1]);context.write(keyPair, value);}}// 自定义reducepublic static class Reduce extends ReducerIntPair, Text, Text, Text {private static final Text SEPARATOR new Text(------------------------------------------------);public void reduce(IntPair key, IterableText values, Context context)throws IOException, InterruptedException {context.write(SEPARATOR, null);for (Text val : values) {context.write(null, val);}}}public static void main(String[] args) throws IOException,InterruptedException, ClassNotFoundException {File jarFile EJob.createTempJar(bin);ClassLoader classLoader EJob.getClassLoader();Thread.currentThread().setContextClassLoader(classLoader);Configuration conf new Configuration(true);String[] otherArgs new String[2];otherArgs[0] hdfs://192.168.1.100:9000/data/test_in/secondary_sort_data_string.txt;String time new SimpleDateFormat(yyyyMMddHHmmss).format(new Date());otherArgs[1] hdfs://192.168.1.100:9000/data/test_out/mr- time;// 实例化一道作业Job job new Job(conf, secondarysort);job.setJarByClass(SecondarySort.class);((JobConf) job.getConfiguration()).setJar(jarFile.toString());// Mapper类型job.setMapperClass(Map.class);// 不再需要Combiner类型因为Combiner的输出类型Text,// IntWritable对Reduce的输入类型IntPair, IntWritable不适用// job.setCombinerClass(Reduce.class);// Reducer类型job.setReducerClass(Reduce.class);// 分区函数job.setPartitionerClass(FirstPartitioner.class);// 分组函数job.setGroupingComparatorClass(GroupingComparator.class);// map 输出Key的类型job.setMapOutputKeyClass(IntPair.class);// map输出Value的类型job.setMapOutputValueClass(Text.class);// rduce输出Key的类型是Text因为使用的OutputFormatClass是TextOutputFormatjob.setOutputKeyClass(Text.class);// rduce输出Value的类型job.setOutputValueClass(Text.class);// 将输入的数据集分割成小数据块splites同时提供一个RecordReder的实现。job.setInputFormatClass(TextInputFormat.class);// 提供一个RecordWriter的实现负责数据输出。job.setOutputFormatClass(TextOutputFormat.class);// 输入hdfs路径FileInputFormat.setInputPaths(job, new Path(otherArgs[0]));// 输出hdfs路径
// FileSystem.get(conf).delete(new Path(args[1]), true);FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));// 提交jobSystem.exit(job.waitForCompletion(true) ? 0 : 1);}
} 3.3 执行结果如下所示 ------------------------------------------------ org ------------------------------------------------hello time------------------------------------------------import javaimport javaimport javaimport javaimport orgimport orgimport orgimport org1import org2------------------------------------------------import1 orgimport1 orgimport1 orgimport1 orgimport1 org------------------------------------------------import2 org1import2 org2import2 org3------------------------------------------------importin org 转https://www.cnblogs.com/minkaihui/p/4125672.html转载于:https://www.cnblogs.com/likanmama/p/7804949.html