• mapreduce的排序機制之total排序

（1）設(shè)置一個reduce task ，全局有序，但是并發(fā)度太低，單節(jié)點負載太大
（2）設(shè)置分區(qū)段partitioner，設(shè)置相應(yīng)數(shù)量的reduce task，可以實現(xiàn)全局有序，但難以避免數(shù)據(jù)分布不均勻——數(shù)據(jù)傾斜問題，有些reduce task負載過大，而有些則過小；
（3）可以通過編寫一個job來統(tǒng)計數(shù)據(jù)分布規(guī)律，獲取合適的區(qū)段劃分，然后用分區(qū)段partitioner來實現(xiàn)排序；但是這樣需要另外編寫一個job對整個數(shù)據(jù)集運算，比較費事
（4）利用hadoop自帶的取樣器，來對數(shù)據(jù)集取樣并劃分區(qū)段，然后利用hadoop自帶的TotalOrderPartitioner分區(qū)來實現(xiàn)全局排序

/*** 全排序示例* @author zhangxueliang**/ public class TotalSort {static class TotalSortMapper extends Mapper<Text, Text, Text, Text> {OrderBean bean = new OrderBean();@Overrideprotected void map(Text key, Text value, Context context) throws IOException, InterruptedException {// String line = value.toString();// String[] fields = line.split("\t");// bean.set(fields[0], Double.parseDouble(fields[1]));context.write(key, value);}}static class TotalSortReducer extends Reducer<Text, Text, Text, Text> {@Overrideprotected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {for (Text v : values) {context.write(key, v);}}}public static void main(String[] args) throws Exception {Configuration conf = new Configuration();Job job = Job.getInstance(conf);job.setJarByClass(TotalSort.class);job.setMapperClass(TotalSortMapper.class);job.setReducerClass(TotalSortReducer.class); // job.setOutputKeyClass(OrderBean.class); // job.setOutputValueClass(NullWritable.class);//用來讀取sequence源文件的輸入組件job.setInputFormatClass(SequenceFileInputFormat.class);FileInputFormat.setInputPaths(job, new Path(args[0]));FileOutputFormat.setOutputPath(job, new Path(args[1]));// job.setPartitionerClass(RangePartitioner.class);//分區(qū)的邏輯使用的hadoop自帶的全局排序分區(qū)組件job.setPartitionerClass(TotalOrderPartitioner.class);//系統(tǒng)自帶的這個抽樣器只能針對sequencefile抽樣RandomSampler randomSampler= new InputSampler.RandomSampler<Text,Text>(0.1,100,10);InputSampler.writePartitionFile(job, randomSampler);//獲取抽樣器所產(chǎn)生的分區(qū)規(guī)劃描述文件Configuration conf2 = job.getConfiguration();String partitionFile = TotalOrderPartitioner.getPartitionFile(conf2);//把分區(qū)描述規(guī)劃文件分發(fā)到每一個task節(jié)點的本地job.addCacheFile(new URI(partitionFile));//設(shè)置若干并發(fā)的reduce taskjob.setNumReduceTasks(3);job.waitForCompletion(true);} }

總結(jié)

以上是生活随笔為你收集整理的Hadoop大数据——mapreduce的排序机制之total排序的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。