hadoop集群算法调用--web平台2.0

问题导读
1、什么是mahout算法？
2、使用mahout算法以及hadoop的MR算法时，流程是怎样的？
2、如何去验证输入的jobtracker 和namenode机器名和端口是否正确？






前记
各种版本：Mahout版本：0.7，hadoop版本：1.0.4，jdk：1.7.0_25 64bit，struts2
项目源码：下载

项目继续沿用之前的框架struts2。当使用mahout算法以及hadoop的MR算法的时候，前台输入参数，action接收，然后启动一个线程，跳转到MR算法监控界面。如果是一般的操作，比如数据的查看以及转换那么前台输入参数后，后台直接处理，然后返回原页面，同时给出提示，不再新开线程。（项目新增了hadoop的基础操作，包括读取、写入、上传、下载和转换等等）。下面的分析，一般是按照先看效果图，然后再分析源码的方式进行。

1.配置
1.1首页
打开浏览器，输入项目发布的首页，即可看到下面的界面：



1.2 配置
配置功能其实就是让用户配置hadoop集群的，配置界面如下：



输入jobtracker 和namenode的机器名和端口号，点击提交，如果成功，就会给出红色提示信息。那么这样如何去验证的呢？
其实就是根据输入的参数得到一个jobclient对象，如果可以得到那么就说明集群是可用的，否则则说明不可用；具体代码如下：

public static boolean initialJobClient(){  
        if(HOST==null||JOBTRACKER_PORT==0||NAMENODE_PORT==0){  
            return false;  
        }  
        log.info("Initial  job client begins...");  
        boolean flag=true;  
        try {  
            InetSocketAddress jobTracker=new InetSocketAddress(HOST,JOBTRACKER_PORT);  
            jobClient=new JobClient(jobTracker, getConf());  
        } catch (IOException e) {  
            flag=false;  
            log.info("Job client can't be got\n"+e.getMessage());  
        }  
        if(flag){  
            log.info("Initial  job client done!!!");  
        }  
        return flag;  
    }  
复制代码

但是，如果当集群不可用的时候，就会一直连接，后台一直打印重试连接，前台一直等待，暂时还没有进行相应的处理。
2. hadoop读取删除
hadoop读取删除其实就是hadoop文件的一般操作，上传下载主要是使用FileSystem做的。
2.1 写入
可以按照下面界面输入相应的参数：


点击提交后，会有红色字体提示写入成功，或者可以在云平台进行数据查看（50070界面）。这里写入使用的是FSDataOutputStream的write功能，这个类有好几个write（不同数据类型，这里使用的是string的），不同的write方式，读取需要使用相应的方式。写入的代码如下：

/** 
     * write string to hdfs file 
     * @param path 
     * @param data 
     * @return 
     */  
    public static boolean writeToHdfs(String path,String data){  
        boolean flag = true;  
        Path filePath=new Path(path);    
        FileSystem fs;    
        FSDataOutputStream out=null;    
        try {    
            fs = FileSystem.get(filePath.toUri(),getConf());  
            out = fs.create(filePath);    
            out.writeUTF(data);    
        } catch(Exception e){    
            log.info("write to hdfs file"+filePath.toString()+" :"+e.getMessage());  
            flag=false;  
        }finally {    
            try {  
                if(out!=null){  
                    out.close();  
                }  
            } catch (IOException e) {  
                log.info("close hdfs file "+filePath.toString()+" wrong\n"+e.getMessage());  
                flag=false;  
            }    
        }   
        return flag;  
    }  
复制代码

2.2 读取
这里读取的还是刚才写入的内容：


输入对应的目录，提交后即可看到内容。需要说明的是这里读取的数据一定要是用writeUTF的方式写入的，不然会是乱码。后面应该会修改这个，改的更加通用点。读取的代码如下：

/** 
     * read hdfs file to String 
     * @param file 
     * @return 
     */  
    public static String readHdfs(String path){  
        String data="";  
        Path filePath= new Path(path);  
        FileSystem fs =null;    
        FSDataInputStream in = null;    
        try {    
          fs=FileSystem.get(filePath.toUri(), getConf());   
          in=fs.open(filePath);  
          data= in.readUTF();    
        }catch(Exception e){  
            log.info("read hdfs file "+path+" error:\n"+e.getMessage());  
        }finally {    
            try {  
                if(in!=null){  
                    in.close();  
                }  
                if(fs!=null){  
                    fs.close();  
                }  
            } catch (IOException e) {  
                log.info("close FileSystem error:\n"+e.getMessage());  
            }  
        }    
        return data;  
    }
复制代码

2.3 上传
上传其实就是使用FileSystem的copy函数而已，主要代码如下：

Path in=new Path(localFile);  
        Path out=new Path(hdfsFile);  
        FileSystem fs=null;  
        try {  
            fs = FileSystem.get(URI.create(hdfsFile),getConf());  
            fs.copyFromLocalFile(in,out);  
复制代码

2.4 下载
下载也是一样，使用的同样是FileSystem的函数，具体如下：

Path in=new Path(hdfsFileName);  
<span style="white-space:pre">      </span>Path out=new Path(localFileName);  
<span style="white-space:pre">      </span>try {  
<span style="white-space:pre">          </span>fs=FileSystem.get(URI.create(hdfsFileName),getConf());  
<span style="white-space:pre">          </span>fs.copyToLocalFile(in, out);  
复制代码

3. hadoop 算法
3.1 序列文件读取
hadoop算法本来是想写成MR的形式的，但是写MR又要上传代码到hadoop集群，感觉麻烦，所以没有写成MR的模式，也就是说没有了监控（其实，后面还是要把代码.class文件打包上传到云平台的）。这里的序列文件读取，其实就是使用了Sequence.Reader而已。需要说明的是，这个读取只是针对一般的<key,value>格式，比如LongWritable、IntWritable等等。读取后的文件一个如下所示：



并且，如果没有序列文件的话，可以按照页面的提示，生成一个序列文件。
3.2 序列文件转换为txt
其实，就是多了一个步骤，把3.1的文件写入一个本地文件而已。写入本地文件使用的是PrintStream，具体如下：

/** 
     * 写入字符串到本地文件 
     * @param localPath 
     * @param data 
     */  
    public static boolean writeToLocal(String localPath,String data){  
        boolean flag=true;  
        File file=new File(localPath);  
        if(!file.exists()){  
            try {  
                new File(file.getParent()).mkdirs();  
                file.createNewFile();  
            } catch (Exception e) {  
                log.info("创建文件失败\n"+e.getMessage());  
            }  
              
        }  
        PrintStream ps=null;  
        try {  
            ps = new PrintStream(new FileOutputStream(file));  
            ps.print(data);  
        } catch (FileNotFoundException e) {  
            log.info("写入本地文件"+localPath+"失败\n"+e.getMessage());  
            flag=false;  
        }finally{  
            ps.close();  
        }  
        return flag;  
    }  
复制代码

4. mahout算法
mahout算法设计的思路其实和前面监控算法的程序是一样的，只是这里在读取云平台已有job列表的时候加了判断，如下：

JobStatus[] jobStatusAll=HadoopUtil.jobClient.getAllJobs();  
        JobStatus jobStatus=null;  
        int id =0;  
        String jobIden="";  
        /** 
         * 防止当前云平台是第一次启动，这个时候没有任务列表，获取的jobStatus是空; 
         */  
        if(jobStatusAll==null||jobStatusAll.length<=0){  
            //修改TaskTracker代码，把集群启动时间写入hdfs，然后在这里读取出来  
            id=0;  
            jobIden=readJTStartTime();  
              
        }else{  
            jobStatus=jobStatusAll[jobStatusAll.length-1];  
            id=jobStatus.getJobID().getId();  
            jobIden=jobStatus.getJobID().getJtIdentifier();  
        }  
复制代码

这里，如果第一次启动的时候，文件列表是空的，这时就需要去读取启动时间，这里采取的方式可以参考：《hadoop 启动时间写入文件》。而这里的代码只是读取文件中的启动时间，然后去拼凑jobId而已。
监控的一般形式如下：


但是，也有可能会出现下面的界面：



看上面两个红色方框，第一个因为前一个任务刚启动然后就失败了，导致jobid没有写入到joblist中。第二个是因为页面刷新的时间和任务启动之间的间隔关系，如果job任务刚刚结束之前读取了一个job状态是running，然后隔了一段时间后，另外一个job任务启动了，那么之前的job任务的状态就会是running的状态，而没有更新。这里的逻辑还有点问题，有待更新。

目前mahout算法可以使用的是Canopy、Kmeans、协同过滤算法，模式挖掘还没有写。
另外需要注意的一点是，工程下载后需要把项目的.class文件打包上传到云平台才不会出错。因为在canopy、kmeans算法的时候，lz自己写了一个数据转换的程序，运行这两个算法的时候会首先调用数据转换，所以没有的话就会报错了。
最后说下数据格式吧：
canopy和kmeans算法的都是一样的，一般如下：

1,133,8  
5,122775,10  
9,18297,6  
9,50422,8  
9,80503,10  
9,110624,8  
9,147283,10  
9,218923,5  
18,102606,2  
25,58155,2  
复制代码

里面的逗号可以是其他字符，需要在算法参数页面设置。
协同过滤算法也是上面的数据格式，不过逗号不能改，且一定是userID,ItemID,prefValue的格式。