並行演算法能算很多東西，不只是計數，wordCount是一個比較簡單的例子，很多其他的請參見我上傳的基於mapreduce 的並行演算法的設計。

今天來實現一個排序的簡單例子。實現過程從簡，因為具體的流程在我寫的wordCount中已經詳細的寫在註釋裡了

首先輸入是一堆檔案file1、file2……裡面存著數字，具體的邏輯是先對數字進行分塊，比如100-200放在一起，200-300……然後每組分別分發給下面，

算完結果一拼就ok了

具體不囉嗦，直接貼程式碼

map

package sort;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;


import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/**
 * Created by zhangguanlong on 2017/11/15.
 */
public class SortMapper extends Mapper<Object, Text, IntWritable, IntWritable>{

    private static IntWritable data = new IntWritable();

    public void map(Object key, Text value, Context context)
            throws IOException, InterruptedException {
        String line = value.toString();

        data.set(Integer.parseInt(line));

        context.write(data, new IntWritable(1));

    }
}
 

package sort;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
/**
 * Created by zhangguanlong on 2017/11/15.
 */
public class SortReducer extends Reducer<IntWritable, IntWritable, IntWritable, IntWritable>{
    private static IntWritable linenum = new IntWritable(1);

    public void reduce(IntWritable key, Iterable<IntWritable> values,
                       Context context) throws IOException, InterruptedException {

        for (IntWritable val : values) {

            context.write(linenum, key);

            linenum = new IntWritable(linenum.get() + 1);
        }

    }
}
 

package sort;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;



import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

/**
 * Created by zhangguanlong on 2017/11/15.
 */
public class SortRunner {
    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;
        }
    }
    /**
     * @param args
     */

    public static void main(String[] args) throws Exception {
        // TODO Auto-generated method stub
        Configuration conf = new Configuration();
        Job job = new Job(conf, "Sort");
        job.setJarByClass(SortRunner.class);
        job.setMapperClass(SortMapper.class);
        job.setPartitionerClass(Partition.class);
        job.setReducerClass(SortReducer.class);
        job.setOutputKeyClass(IntWritable.class);
        job.setOutputValueClass(IntWritable.class);
        FileInputFormat.addInputPath(job, new Path("/wc/sort1/"));
        FileOutputFormat.setOutputPath(job, new Path("/wc/sort2/"));
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}
 

注意這裡的file本來是放在linux根目錄下的，不懂得可以去看hadoop 和linux的shell指令，然後執行

[[email protected] ~]$ hadoop jar SProject.jar sort.SortRunner

成功跑起來是這個樣子

還要注意的是檔案裡一定不要有空格，內容必須是數字，因為這只是個簡單的demo。。。。

跑完看下結果

over