Hadoop好友推荐系统-执行聚类算法

灰太狼 2022-06-12 01:51 262阅读 0赞

### 项目总目录：[基于Hadoop的好友推荐系统项目综述][Hadoop] ###

## 一、执行聚类算法 ##

### 1、前端展示 ###

*  jsp页面

<form id="formid" method="post">
                <table>
                    <tr>
                        <td><label for="name">输入路径:</label></td>
                        <td><input class="easyui-validatebox" type="text"
                            name="input" data-options="required:true" 
                            style="width:200px" value="/user/root/_filter/caldistance" />
                        </td>
    
                    </tr>
    
                    <tr>
                        <td><label for="name">距离阈值:</label></td>
                        <td><input class="easyui-validatebox" type="text"
                            id="delta" name="delta" data-options="required:true"  />
                        </td>
    
                    </tr>
                    <tr>
                        <td><label for="name">密度计算方法:</label></td>
                        <td>
                        <select id="method" class="easyui-combobox" name="method"
                            style="width:200px;">
                                <option value="cutoff">Cut-off算法</option>
                                <option value="gaussian">Gaussian算法</option>
                            </select>
                        </td>
                    </tr>
    
                    <tr>
                        <td><label for="name">局部密度MR Reducer个数:</label></td>
                        <td><input class="easyui-validatebox" type="text"
                             name="numReducerDensity" data-options="required:true" value="1" />
                        </td>
                    </tr>
                    <tr>
                        <td><label for="name">最小距离MR Reducer个数:</label></td>
                        <td><input class="easyui-validatebox" type="text"
                             name="numReducerDistance" data-options="required:true" value="1" />
                        </td>
                    </tr>
                    <tr>
                        <td><label for="name">排序MR Reducer个数(支持设置1个):</label></td>
                        <td><input class="easyui-validatebox" type="text"
                             name="numReducerSort" data-options="required:true" value="1" />
                        </td>
                    </tr>
                    <tr>
                        <td><a id="submitid" href="#"
                            class="easyui-linkbutton">提交</a></td>
                        <td><a id="resetid" href="#"
                            class="easyui-linkbutton">重置</a></td>
                    </tr>
                </table>
            </form>

执行聚类算法包括三个MR任务：计算局部密度MR、计算最小距离MR以及结果排序MR任务。需要设置如下参数：  
输入路径：默认就是距离计算任务的输出目录；  
距离阀值：手工设置，可以参考计算最佳DC任务的结果值，但是效果未必好。这里根据多次试验，选择100作为阀值；  
密度计算方法：可以选择cutoff和gaussian两种计算方法，默认使用cutoff方法；  
局部密度MR Reducer个数：默认1个；  
最小距离MR Reducer个数：默认1个；  
排序MR Reducer个数：只支持设为1个。

*  js逻辑

$('#submitid').bind('click', function() {
        
            console.info("submitid提交！");
            $('#formid').form({
                url : 'cloud/cloud_runCluster1.action', // 提交runCluster1任务，跳转到监控页面
                onSubmit : function() {
        
                    // do some check
                    // return false to prevent submit;
                },
                success : function(data) {
        
                    // 关闭弹窗
                    closeProgressbar();
    
                    if (data == 'fail') {
       // cf算法Thread创建失败
                        $.messager.alert('提示', '未能提交RunCluster1算法到云平台，请检查后台日志！');
                    } else {
       // 已提交任务，转到监控页面
    
                        // 使用单独Tab的方式
                        layout_center_addTabFun({
                            title : 'MR算法监控',
                            closable : true,
                            // iconCls : node.iconCls,
                            href : 'cluster/monitor_one.jsp' //监控页面
                        });
                    }
                }
            });
            // submit the form
            $('#formid').submit();
            // 弹出progressbar
            popupProgressbar('Please waiting', '提交RunCluster1算法任务中...', '2000');
        });
    
        // connectclusterreset button
        $('#resetid').bind('click', function() {
        
            console.info("resetid重置！");
            $('#formid').form('reset', {});
        });

### 后台逻辑 ###

*  action层

对应的action从这里获取：url : ‘cloud/cloud\_runCluster1.action’。

/**
         * 执行聚类
         */
        public void runCluster1(){  
            Map<String ,Object> map = new HashMap<String,Object>();
            try {
                HUtils.setJobStartTime(System.currentTimeMillis()-10000);//设置任务启动时间
                HUtils.JOBNUM=3;//设置任务数
                // 使用Thread的方式启动一组MR任务
                new Thread(new RunCluster1(input, delta, method,numReducerDensity,
                        numReducerDistance,numReducerSort)).start();
                // 启动成功后，直接返回到监控，同时监控定时向后台获取数据，并在前台展示；
    
                map.put("flag", "true");//任务启动成功
                map.put("monitor", "true");//任务监控开始
            } catch (Exception e) {
                e.printStackTrace();
                map.put("flag", "false");
                map.put("monitor", "false");
                map.put("msg", e.getMessage());
            }
            Utils.write2PrintWriter(JSON.toJSONString(map));
        }

其中的关键语句是

new Thread(new RunCluster1(input, delta, method,numReducerDensity,
                        numReducerDistance,numReducerSort)).start();

该语句通过启动线程启动MapReduce任务。  
RunCluster1的定义如下：

/**
     * 1. 计算局部密度
     * 2. 计算最小距离
     * 3. 排序
     */
    public class RunCluster1 implements Runnable {
        
    
        private String input;//输入路径
        private String numReducerDensity;//局部密度MR reduce数目
        private String numReducerDistance;//最短距离MR reduce数目
        private String numReducerSort;//排序MR reduce数目
        private String dc;//距离阀值
        private String method;//密度计算方法
    
        public RunCluster1(String input,String dc,String method,String numReducerDensity,
                String numReducerDistance,String numReducerSort){
            this.input=input;
            this.dc=dc;
            this.method=method;
            this.numReducerDensity=numReducerDensity;
            this.numReducerDistance=numReducerDistance;
            this.numReducerSort=numReducerSort;
        }
        @Override
        public void run() {
            String [] args =new String[]{
                    HUtils.getHDFSPath(input),
                    HUtils.getHDFSPath(HUtils.LOCALDENSITYOUTPUT),
                    dc,
                    method,
                    numReducerDensity
            };//初始化计算局部密度MR的参数
            try {
                int ret=
                ToolRunner.run(HUtils.getConf(), new LocalDensityJob(),args );//计算局部密度
                if(ret!=0){
                    Utils.simpleLog("LocalDensityJob任务运行失败！");
                    return ;
                }
                Thread.sleep(3000);// 等待3秒时间
                args=new String[]{
                        HUtils.getHDFSPath(input),// 使用距离计算后的路径作为输入
                        HUtils.getHDFSPath(HUtils.DELTADISTANCEOUTPUT),
                        HUtils.getHDFSPath(HUtils.LOCALDENSITYOUTPUT),
                        numReducerDistance
                };//初始化计算最短距离MR的参数
    
                ret=ToolRunner.run(HUtils.getConf(), new DeltaDistanceJob(), args);//计算最短距离
                if(ret!=0){
                    Utils.simpleLog("DeltaDistanceJob任务运行失败！");
                    return ;
                }
    
                Thread.sleep(3000);// 等待3秒时间
    
                args=new String[]{
                        HUtils.getHDFSPath(HUtils.DELTADISTANCEOUTPUT),
                        HUtils.getHDFSPath(HUtils.SORTOUTPUT),
                        numReducerSort
                };//初始化排序MR的参数
                ret=ToolRunner.run(HUtils.getConf(), new SortJob(), args);//排序
                if(ret!=0){
                    Utils.simpleLog("SortJob任务运行失败！");
                    return ;
                }
    
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    //省略了get/set方法
    }

其中的关键语句如下：

ToolRunner.run(HUtils.getConf(), new LocalDensityJob(),args );//计算局部密度
    ret=ToolRunner.run(HUtils.getConf(), new DeltaDistanceJob(), args);//计算最短距离
    ret=ToolRunner.run(HUtils.getConf(), new SortJob(), args);//排序

*  计算局部密度

LocalDensityJob定义如下：

/**
     * 
     * 输入为 <key,value>--> <distance,<id_i,id_j>>
     *  <距离，<向量i编号，向量j编号>>
     *  
     *  Mapper：
     *  输出向量i编号，1
     *      向量j编号,1
     *  Reducer:
     *  输出
     *     向量i编号，局部密度
     *     有些向量是没有局部密度的，当某个向量距离其他点的距离全部都大于给定阈值dc时就会发生
     */
    public class LocalDensityJob extends Configured implements Tool {
        
    
        /**
         * @param args
         * @throws Exception
         * method: gaussian,cutoff
         */
        public static void main(String[] args) throws Exception {
            ToolRunner.run(HUtils.getConf(), new LocalDensityJob(), args);
        }
    
        public int run(String[] args) throws Exception {
    
            Configuration conf = HUtils.getConf();
            String[] otherArgs = new GenericOptionsParser(conf, args)
                    .getRemainingArgs();//初始化命令行参数
            if (otherArgs.length != 5) {
       //要求参数个数必须符合要求
                System.err
                        .println("Usage: com.kang.fast_cluster.LocalDensityJob <in> <out> <dc> <method> <num_reducer>");
                System.exit(4);
            }
            conf.set("METHOD", otherArgs[3]);//设置密度计算方法
            conf.setDouble("DC", Double.parseDouble(otherArgs[2]));//设置距离阀值
            Job job = Job.getInstance(conf,
                    "Count the near neighours with a given dc(distance):"
                            + otherArgs[2]);
            job.setJarByClass(LocalDensityJob.class);
            job.setMapperClass(LocalDensityMapper.class);
            job.setCombinerClass(LocalDensityReducer.class);
            job.setReducerClass(LocalDensityReducer.class);
    
            job.setNumReduceTasks(Integer.parseInt(otherArgs[4]));//设置reduce个数
    
            job.setOutputKeyClass(IntWritable.class);
            job.setOutputValueClass(DoubleWritable.class);
    
            job.setOutputFormatClass(SequenceFileOutputFormat.class);
            job.setInputFormatClass(SequenceFileInputFormat.class);
    
            SequenceFileInputFormat.addInputPath(job, new Path(otherArgs[0]));
    
            SequenceFileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
            FileSystem.get(conf).delete(new Path(otherArgs[1]), true);
            return job.waitForCompletion(true) ? 0 : 1;
    
        }
    
    }

map方法定义如下：

/**
     * 输入为<距离d_ij,<向量i编号，向量j编号>>
     * 根据距离dc阈值判断距离d_ij是否小于dc，符合要求则
     * 输出
     * 向量i编号，1
     * 向量j编号，1
     */
    public class LocalDensityMapper extends Mapper<DoubleWritable, IntPairWritable, IntWritable, DoubleWritable> {
    
        private double dc;
        private String method =null;
    
        private IntWritable vectorId= new IntWritable();
        private DoubleWritable one= new DoubleWritable(1);
    
        @Override 
        public void setup(Context cxt){
            dc=cxt.getConfiguration().getDouble("DC", 0);
            method = cxt.getConfiguration().get("METHOD", "gaussian");
        }
    
        @Override
        public void map(DoubleWritable key,IntPairWritable value,Context cxt)throws InterruptedException,IOException{
            double distance= key.get();
    
            if(method.equals("gaussian")){
                one.set(Math.pow(Math.E, -(distance/dc)*(distance/dc)));//gaussian计算局部密度
            }
    
            if(distance<dc){
       //两者距离小于阀值，则输出
                vectorId.set(value.getFirst());
                cxt.write(vectorId, one);
                vectorId.set(value.getSecond());
                cxt.write(vectorId, one);
            }
        }
    
    
    
    }

combiner和reduce方法定义如下：

/**
     * map端的输出格式为<i,list(1,1,1---)>,i为用户id号
     * reduce任务就是把list(1,1,1---)中的所有“1”相加即可。
     */
    public class LocalDensityReducer extends
            Reducer<IntWritable, DoubleWritable, IntWritable, DoubleWritable> {
        private DoubleWritable sumAll = new DoubleWritable();
    
        @Override
        public void reduce(IntWritable key, Iterable<DoubleWritable> values,Context cxt)
        throws IOException,InterruptedException{
            double sum =0;
            for(DoubleWritable v:values){
                sum+=v.get();//每遍历一次加1
            }
            sumAll.set(sum);// 统计总数
            cxt.write(key, sumAll);
            Utils.simpleLog("vectorI:"+key.get()+",density:"+sumAll);
        }
    }

*  计算最小距离

DeltaDistanceJob定义如下：

/**
     * 寻找大于自身密度的最小其他向量的距离
     * mapper输入：
     * 输入为<距离d_ij,<向量i编号，向量j编号>>
     * 把LocalDensityJob的输出
     *      i,density_i
     * 放入一个map中，用于在mapper中进行判断两个局部密度的大小以决定是否输出
     * mapper输出：
     *      i,<density_i,min_distance_j>
     *      IntWritable,DoublePairWritable
     * reducer 输出：
     *      <density_i*min_distancd_j> <density_i,min_distance_j,i>
     *      DoubleWritable,  IntDoublePairWritable
     */
    public class DeltaDistanceJob extends Configured implements Tool {
        
    
        /**
         * @param args
         * @throws Exception 
         */
        public static void main(String[] args) throws Exception {
            ToolRunner.run(HUtils.getConf(), new DeltaDistanceJob(), args);
        }
    
        public int run(String[] args) throws Exception {
    
            Configuration conf = HUtils.getConf();
            String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();//初始化命令行参数
            if (otherArgs.length !=4) {
       //要求参数个数必须为4
              System.err.println("Usage: com.kang.fast_cluster.DeltaDistanceJob <in> <out> <local_density> <num_reducer>");
              System.exit(4);
            }
            conf.set("DENSITYPATH", otherArgs[2]);//指定局部密度文件存放的目录
            Job job =  Job.getInstance(conf,"find the nearest distance with the bigger near neighours");
            job.setJarByClass(DeltaDistanceJob.class);
            job.setMapperClass(DeltaDistanceMapper.class);
    //      job.setCombinerClass(DeltaDistanceReducer.class);
            job.setReducerClass(DeltaDistanceReducer.class);
    
            job.setNumReduceTasks(Integer.parseInt(otherArgs[3]));//指定reduce数目
    
            job.setMapOutputKeyClass(IntWritable.class);
            job.setMapOutputValueClass(DoublePairWritable.class);
    
            job.setOutputKeyClass(DoubleWritable.class);
            job.setOutputValueClass(IntDoublePairWritable.class);
    
            job.setInputFormatClass(SequenceFileInputFormat.class);
            job.setOutputFormatClass(SequenceFileOutputFormat.class);
            SequenceFileInputFormat.addInputPath(job, new Path(otherArgs[0]));
            SequenceFileOutputFormat.setOutputPath(job,new Path(otherArgs[1]));
    
            FileSystem.get(conf).delete(new Path(otherArgs[1]), true);
            return job.waitForCompletion(true) ? 0 : 1;
    
        }
    }

map方法实现如下：

**
     * 计算最小距离mapper方法
     * map的输出为<i,(局部密度，i和j的距离)>
     */
    public class DeltaDistanceMapper extends
            Mapper<DoubleWritable, IntPairWritable, IntWritable, DoublePairWritable> {
        private Logger log = LoggerFactory.getLogger(DeltaDistanceMapper.class);
    
        private IntWritable vector_i= new IntWritable();
        private DoublePairWritable density_distance = new DoublePairWritable();
    
        private Path densityPath ;
        private Map<Integer,Double> densityMap= new HashMap<Integer,Double>();// vector_id,density
        private int max_density_vector_id=-1;// 最大的局部密度下标
        private double max_density=-Double.MAX_VALUE;// 最大的局部密度 
        @Override
        public void setup(Context cxt) throws IOException{
            densityPath = new Path(cxt.getConfiguration().get("DENSITYPATH"));//获取局部密度文件的存储目录
            Configuration conf = cxt.getConfiguration();
            SequenceFile.Reader reader = null;
            FileStatus[] fss=densityPath.getFileSystem(conf).listStatus(densityPath);
            for(FileStatus f:fss){
       //遍历存储局部密度文件目录下的文件
                if(!f.toString().contains("part")){
                    continue; // 排除其他文件，只处理文件名包含“part”的文件
                }
                try {
                    reader = new SequenceFile.Reader(conf, Reader.file(f.getPath()),
                            Reader.bufferSize(4096), Reader.start(0));
                    IntWritable dkey = (IntWritable) ReflectionUtils.newInstance(
                            reader.getKeyClass(), conf);
                    DoubleWritable dvalue = (DoubleWritable) ReflectionUtils.newInstance(
                            reader.getValueClass(), conf);
                    while (reader.next(dkey, dvalue)) {
       // 循环读取文件
                        densityMap.put(dkey.get(), dvalue.get());
                        if(dvalue.get()>max_density){
       // 寻找最大局部密度的下标
                            max_density=dvalue.get();//保存当前最大局部密度
                            max_density_vector_id=dkey.get();//保存当前最大局部密度的下标
                        }
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                    max_density_vector_id=-1;
                } finally {
                    IOUtils.closeStream(reader);
                }
            }
            if(max_density_vector_id==-1){
                log.info("没有找到局部密度最大的向量，程序出错！");
            }
            // 把max_density_vector_id写入文件
            FileSystem fs = FileSystem.get(cxt.getConfiguration());
            Path path = new Path(HUtils.DELTADISTANCEBIN);
            if(fs.exists(path)){
                fs.delete(path, false);
            }
            FSDataOutputStream out = fs.create(path);
            try{
                out.writeInt(max_density_vector_id);
            }catch(Exception e){
                e.printStackTrace();
            }finally{
                out.close();
            }
            log.info("densityMap中的键值对个数：{},局部密度最大向量id：{},局部密度：{}",
                    new Object[]{densityMap.size(),max_density_vector_id,max_density});
        }
    
        // distance_ij,<i,j>
        @Override
        public void map(DoubleWritable key, IntPairWritable value, Context cxt) throws IOException,InterruptedException{
            int vectorI= value.getFirst();
            int vectorJ= value.getSecond();
            if(vectorI==max_density_vector_id||vectorJ==max_density_vector_id){
       // 最大局部密度，需寻找离该点最大的距离,应该在reducer中判断
                // 如果是局部密度最大点，应该直接输出即可
                vector_i.set(max_density_vector_id);
                density_distance.setFirst(max_density);
                density_distance.setSecond(key.get());
                cxt.write(vector_i, density_distance);
                log.info("vector_i:{},density:{},distance:{}",new Object[]{max_density_vector_id,max_density,key.get()});
                return ;
    
            }
            // 不是局部密度最大点，则找最小的即可
            double densityI=0;
            double densityJ=0;
            if(!densityMap.containsKey(vectorI)||!densityMap.containsKey(vectorJ)){
       //  两个任一个不存在，则不输出任何
                return ;
            }
    
            densityI=densityMap.get(vectorJ);
            densityJ=densityMap.get(vectorI);
    
            if(densityI<densityJ){
       // 输出   <I,大于I 的density的distance>
                vector_i.set(vectorI);
                density_distance.setFirst(densityI);
                density_distance.setSecond(key.get());// second 是distance
                cxt.write(vector_i, density_distance);
                log.info("vector_i:{},density:{},distance:{}",new Object[]{vectorI,densityI,key.get()});
            }
            if(densityI>densityJ){
       // 输出   <J,大于J 的density的distance>
                vector_i.set(vectorJ);
                density_distance.setFirst(densityJ);
                density_distance.setSecond(key.get());
                cxt.write(vector_i, density_distance);
                log.info("vector_i:{},density:{},distance:{}",new Object[]{vectorJ,densityJ,key.get()});
            }
        }
    }

reduce方法实现如下：

/**
     * 输出
     * <density_i*min_distancd_j> <first:density_i,second:min_distance_j,third:i>
     *      DoubleWritable,  IntDoublePairWritable
     */
    public class DeltaDistanceReducer extends
            Reducer<IntWritable, DoublePairWritable, DoubleWritable, IntDoublePairWritable> {
    
        private IntDoublePairWritable i_density_distance = new IntDoublePairWritable();
        private DoubleWritable mul = new DoubleWritable();
    
        private int max_density_vector_id=-1;
    
        @Override
        public void setup(Context cxt) throws IOException{
            Path path = new Path(HUtils.DELTADISTANCEBIN);
            FileSystem fs = FileSystem.get(cxt.getConfiguration());
            FSDataInputStream in = fs.open(path);
            // 读取最大局部密度
            try{
                max_density_vector_id = in.readInt();
            }catch(IOException e){
                e.printStackTrace();
            }finally{
                in.close();
            }
            Utils.simpleLog("最大局部密度向量id："+max_density_vector_id);
        }
    
        @Override
        public void reduce(IntWritable key,Iterable<DoublePairWritable> values,Context cxt) throws IOException,InterruptedException{
            // 如果是最大局部密度，则寻找最大距离，否则寻找最小距离
            double minDistance = key.get()!=max_density_vector_id?Double.MAX_VALUE:-Double.MAX_VALUE;
            double density=0.0;
            for(DoublePairWritable s:values){
    
                if(key.get()!=max_density_vector_id&&s.getSecond()<minDistance){
       // 寻找距离最小的
                    minDistance = s.getSecond();
                    density=s.getFirst();
                }
                if(key.get()==max_density_vector_id&&s.getSecond()>minDistance){
       // 寻找距离最大的
                    minDistance = s.getSecond();
                    density=s.getFirst();
                }
            }
    
            i_density_distance.setFirst(density);
            i_density_distance.setSecond(minDistance);
            i_density_distance.setThird(key.get());
    
            mul.set(density*minDistance);
            cxt.write(mul, i_density_distance);
            Utils.simpleLog("vectorI:"+key.get()+",density:"+density+",distance:"+minDistance+",mul:"+mul);
        }
    }

*  排序

排序的MapReduce实现非常简单，关键是key值的类型类要实现WritableComparable接口。

/**
     * 对DeltaDistance结果排序
     * 
     * 输入为 <density_i*min_distancd_j> <first:density_i,second:min_distance_j,third:i>
     *      DoubleWritable,  IntDoublePairWritable
     *  
     *  输出是同样的内容，只是做了排序
     */
    public class SortJob extends Configured implements Tool {
        
    
        /**
         * @param args
         * @throws Exception
         * method: gaussian,cutoff
         */
        public static void main(String[] args) throws Exception {
            ToolRunner.run(HUtils.getConf(), new SortJob(), args);
        }
    
        public int run(String[] args) throws Exception {
    
            Configuration conf = HUtils.getConf();
            String[] otherArgs = new GenericOptionsParser(conf, args)
                    .getRemainingArgs();//初始化命令参数
            if (otherArgs.length != 3) {
                System.err
                        .println("Usage: com.kang.fastcluster.SortJob <in> <out> <num_reducer>");
                System.exit(4);
            }
    
            Job job = Job.getInstance(conf,
                    "sort the input:"
                            + otherArgs[0]);
            job.setJarByClass(SortJob.class);
            job.setMapperClass(SortMapper.class);
    //      job.setCombinerClass(SortReducer.class);
            job.setReducerClass(SortReducer.class);
    
            job.setNumReduceTasks(Integer.parseInt(otherArgs[2]));//设置reduce数目
    
            job.setOutputKeyClass(CustomDoubleWritable.class);
            job.setOutputValueClass(IntDoublePairWritable.class);
    
            job.setOutputFormatClass(SequenceFileOutputFormat.class);
            job.setInputFormatClass(SequenceFileInputFormat.class);
    
            SequenceFileInputFormat.addInputPath(job, new Path(otherArgs[0]));
    
            SequenceFileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
            FileSystem.get(conf).delete(new Path(otherArgs[1]), true);
            return job.waitForCompletion(true) ? 0 : 1;
    
        }
    
        private static class SortMapper extends Mapper<DoubleWritable,IntDoublePairWritable,
            CustomDoubleWritable,IntDoublePairWritable>{
            private CustomDoubleWritable mul = new CustomDoubleWritable();
            public void map(DoubleWritable key,IntDoublePairWritable value,Context cxt)throws IOException,
                InterruptedException{
                mul.set(key.get());//将DoubleWritable转为CustomDoubleWritable，用于排序
                //CustomDoubleWritable实现了WritableComparable接口，可以按照其实际数值大小来排序
                cxt.write(mul, value);
            }
        }
    
        private static class SortReducer extends Reducer<CustomDoubleWritable,IntDoublePairWritable,
            CustomDoubleWritable,IntDoublePairWritable>{
            public void reduce(CustomDoubleWritable key,Iterable<IntDoublePairWritable> vs,Context cxt)
                throws IOException,InterruptedException{
                for(IntDoublePairWritable v:vs){
                    cxt.write(key, v);//直接输出即可，MapReduce框架自动为我们排好序
                    Utils.simpleLog("mul:"+key.get()+",other:"+v.toString());
                }
            }
        }
    
    }

## 二、程序运行截图 ##

### 1、前端（距离阀值取100，密度计算方法选cut-off算法） ###

![这里写图片描述][20170712161847955]

本次任务提交一共需要执行三个MapReduce任务，如下：  
![这里写图片描述][20170712162255202]  
![这里写图片描述][20170712162314729]  
![这里写图片描述][20170712162330348]

### 2、后台 ###

![这里写图片描述][20170712162520213]  
![这里写图片描述][20170712162536650]  
![这里写图片描述][20170712162551502]

[Hadoop]: http://blog.csdn.net/xiaokang123456kao/article/details/75006624
[20170712161847955]: /images/20220612/206b681219dd4f188af01e3d97474f6b.png
[20170712162255202]: /images/20220612/57ff13b9621540d89eda077c745dc3ec.png
[20170712162314729]: /images/20220612/e4ed4395ae294e838f16268394e58bfb.png
[20170712162330348]: /images/20220612/7ae746a1302f4eca8948f8b209ff7546.png
[20170712162520213]: /images/20220612/fa149e5270b344d49d0c6632dc3d2368.png
[20170712162536650]: /images/20220612/ffbe90016bca4d7db319412f6bf1badc.png
[20170712162551502]: /images/20220612/8203711ea11644a4ab63495e6e700b50.png