mapreduce排序(自定义Partition) hadoopmapreducePartition
程序员文章站
2022-05-25 09:39:56
...
import java.io.IOException; 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.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; /** * mapreduce排序 * * @author asheng file1:1.txt 2 32 654 32 15 756 65223 * * file2:2.txt 5956 22 650 92 * * file3:3.txt 26 54 6 * 对file1,file2,file3进行排序,能够第一想到的便是mapreduce自动排序,但是这里面有问题: * Reduce排序只是对发送到自己所在的节点的数据进行排序,不能保证整体的顺序 * 所以这里要自定义Partition,保证Partition后,Reduce上的数据在整体上是有序的,然后在reduce内进行排序 */ public class Sort { public static class Map extends Mapper<Object, Text, IntWritable, IntWritable> { private IntWritable data = new IntWritable(); public void map(Object key, Text value, Context context) throws IOException, InterruptedException { data.set(Integer.parseInt(value.toString())); context.write(data, new IntWritable(1)); } } public static class Reduce extends Reducer<IntWritable, IntWritable, IntWritable, IntWritable> { private IntWritable data = new IntWritable(1); public void reduce(IntWritable key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { for (IntWritable v : values) { System.out.println(v); context.write(data, key); data = new IntWritable(data.get() + 1); } } } public static class Partition extends Partitioner<IntWritable, IntWritable> { @Override public int getPartition(IntWritable key, IntWritable value, int numPartitions) { int Maxnumber = 65223; int bound = Maxnumber / numPartitions + 1; int keynumber = key.get(); for (int i = 0; i < numPartitions; i++) { if (keynumber < bound * i && keynumber >= bound * (i - 1)) { return i - 1; } } return 0; } } public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException { Configuration conf = new Configuration(); Job job = new Job(conf, "sort"); job.setJarByClass(Sort.class); job.setMapperClass(Map.class); job.setReducerClass(Reduce.class); job.setPartitionerClass(Partition.class); job.setOutputKeyClass(IntWritable.class); job.setOutputValueClass(IntWritable.class); FileInputFormat.setInputPaths(job, "/home/asheng/hadoop/in"); FileOutputFormat .setOutputPath(job, new Path("/home/asheng/hadoop/out")); job.waitForCompletion(true); } }
推荐阅读
-
Hadoop学习之路(7)MapReduce自定义排序
-
「 Hadoop」mapreduce对温度数据进行自定义排序、分组、分区等 【转载】
-
mapreduce排序中java.io.IOException: Illegal partition for 2 (-1) hadoopmapreducepartitionIOException
-
mapreduce排序(自定义Partition) hadoopmapreducePartition
-
Hadoop学习之路(7)MapReduce自定义排序
-
mapreduce对温度数据进行自定义排序、分组、分区等
-
「 Hadoop」mapreduce对温度数据进行自定义排序、分组、分区等 【转载】