Shulou(Shulou.com)05/31 Report--

This article mainly shows you the "sample analysis of Hadoop-assisted sorting", which is easy to understand and clear. I hope it can help you solve your doubts. Let the editor lead you to study and study the "sample analysis of Hadoop-assisted sorting".

1. Demand

Seek the highest temperature of the year

two。 Sample data

1995 101996 111995 161995 221996 261995 31996 71996 101996 201996 331995 211996 91995 311995-131995 221997-21997 281997 151995 8

3. Ideas, codes

If the records are grouped by year and sorted in descending order of temperature, and then all records for the same year are sent to a reducer group, the first record of each group is the highest temperature for that year. The main points to implement this scheme are:

a. The definition includes a combination of natural keys (year) and natural values (temperature).

b. Sort the records according to the key combination.

c. Only natural keys are considered when partitioning and grouping key combinations.

Import org.apache.hadoop.io.IntWritable;import org.apache.hadoop.io.WritableComparable;import java.io.DataInput;import java.io.DataOutput;import java.io.IOException;/** * key combination, which in this case is used to assist sorting, including year and temperature. * / public class IntPair implements WritableComparable {private IntWritable first; private IntWritable second; public IntPair () {this.first = new IntWritable (); this.second = new IntWritable (); / / if the above two lines are commented out, an exception java.lang.NullPointerException at IntPair.readFields} public IntPair (int first, int second) {set (new IntWritable (first), new IntWritable (second)) will occur. } public IntPair (IntWritable first, IntWritable second) {set (first, second);} public void set (IntWritable first, IntWritable second) {this.first = first; this.second = second;} public IntWritable getFirst () {return first;} public IntWritable getSecond () {return second;} public void write (DataOutput out) throws IOException {first.write (out) Second.write (out);} public void readFields (DataInput in) throws IOException {first.readFields (in); second.readFields (in);} @ Override public int hashCode () {return first.hashCode () * 163+ second.hashCode ();} @ Override public boolean equals (Object obj) {if (obj instanceof IntPair) {IntPair ip = (IntPair) obj Return first.get () = = ip.first.get () & & second.get () = = ip.second.get ();} return false;} @ Override public String toString () {return first + "\ t" + second;} public int compareTo (IntPair o) {int cmp = first.compareTo (o.first) If (cmp = = 0) {cmp = second.compareTo (o.second);} return cmp;}}

Import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.conf.Configured;import org.apache.hadoop.fs.Path;import org.apache.hadoop.io.IntWritable;import org.apache.hadoop.io.LongWritable;import org.apache.hadoop.io.NullWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.io.WritableComparable;import org.apache.hadoop.io.WritableComparator;import org.apache.hadoop.io.WritableUtils;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.output.FileOutputFormat;import org.apache.hadoop.util.GenericOptionsParser;import org.apache.hadoop.util.Tool;import org.apache.hadoop.util.ToolRunner;import java.io.IOException Public class MaxTemperatureUsingSecondarySort extends Configured implements Tool {static class MaxTemperatureMapper extends Mapper {@ Override protected void map (LongWritable key, Text value, Context context) throws IOException, InterruptedException {String [] val = value.toString (). Split ("\\ t"); if (val.length = = 2) {context.write (new IntPair (val [0]), Integer.parseInt (val [1])), NullWritable.get ()) } static class MaxTemperatureReducer extends Reducer {@ Override protected void reduce (IntPair key, Iterable values, Context context) throws IOException, InterruptedException {context.write (key, NullWritable.get ()) / / output only the first line}} / / only based on the first partition public static class FirstPartitioner extends Partitioner {@ Override public int getPartition (IntPair key, NullWritable value, int numPartitions) {return (key.getFirst (). HashCode () & Integer.MAX_VALUE)% numPartitions }} / / grouping only according to first public static class GroupComparator extends WritableComparator {private static final IntWritable.Comparator INT_COMPARATOR = new IntWritable.Comparator (); protected GroupComparator () {super (IntPair.class, true) } @ Override public int compare (byte [] b1, int S1, int L1, byte [] b2, int S2, int L2) {try {int firstL1 = WritableUtils.decodeVIntSize (b1 [S1]) + readVInt (b1, S1); int firstL2 = WritableUtils.decodeVIntSize (b2 [S2]) + readVInt (b2, S2) Return INT_COMPARATOR.compare (b1, S1, firstL1, b2, S2, firstL2);} catch (IOException e) {throw new IllegalArgumentException (e) } @ Override public int compare (WritableComparable a, WritableComparable b) {if (an instanceof IntPair & & b instanceof IntPair) {return ((IntPair) a) .getFirst () .compareTo (IntPair) b) .getFirst ());} return super.compare (a, b) }} / / sort public static class KeyComparator extends WritableComparator {protected KeyComparator () {super (IntPair.class, true);} @ Override public int compare (WritableComparable a, WritableComparable b) {if (an instanceof IntPair & & b instanceof IntPair) {IntPair ip1 = (IntPair) a; IntPair ip2 = (IntPair) b by key combination Int cmp = ip1.getFirst () .compareTo (ip2.getFirst ()); / / Ascending (year) if (cmp! = 0) {return cmp;} return-ip1.getSecond () .compareTo (ip2.getSecond ()); / / descending (temperature)} return super.compare (a, b) } public int run (String [] args) throws Exception {Configuration conf = new Configuration (); String [] otherArgs = new GenericOptionsParser (conf, args). GetRemainingArgs (); if (otherArgs.length! = 2) {System.err.println ("Parameter number is wrong, please enter two parameters:"); System.exit (- 1);} Path inputPath = new Path (otherArgs [0]) Path outputPath = new Path (otherArgs [1]); / / conf.set ("fs.defaultFS", "hdfs://vmnode.zhch:9000"); Job job = Job.getInstance (conf, "MaxTemperatureUsingSecondarySort"); / / job.setJar ("F:/workspace/AssistRanking2/target/AssistRanking2-1.0-SNAPSHOT.jar"); job.setJarByClass (MaxTemperatureUsingSecondarySort.class); job.setMapperClass (MaxTemperatureMapper.class) Job.setPartitionerClass (FirstPartitioner.class); job.setSortComparatorClass (KeyComparator.class); / / sort job.setGroupingComparatorClass (GroupComparator.class) by default according to Key's compareTo function; job.setReducerClass (MaxTemperatureReducer.class); job.setMapOutputKeyClass (IntPair.class); job.setOutputKeyClass (IntPair.class); job.setOutputValueClass (NullWritable.class); FileInputFormat.addInputPath (job, inputPath); FileOutputFormat.setOutputPath (job, outputPath) Return job.waitForCompletion (true)? 0: 1;} public static void main (String [] args) throws Exception {int exitCode = ToolRunner.run (new MaxTemperatureUsingSecondarySort (), args); System.exit (exitCode);}}

4. Run screenshot

The above is all the contents of the article "sample Analysis of Hadoop-assisted sorting". Thank you for reading! I believe we all have a certain understanding, hope to share the content to help you, if you want to learn more knowledge, welcome to follow the industry information channel!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.