客户端用java api 远程操作HDFS以及远程提交MR任务(源码和异常处理)

package mapreduce;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hdfs.DistributedFileSystem;
import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
import org.apache.hadoop.io.IOUtils;
/**
 * file operation on HDFS
 * @author liuxingjiaofu
 *
 */
public class HDFS_File {
    //read the file from HDFS
    public void ReadFile(Configuration conf, String FileName){
      try{
            FileSystem hdfs = FileSystem.get(conf);
            FSDataInputStream dis = hdfs.open(new Path(FileName));
            IOUtils.copyBytes(dis, System.out, 4096, false); 
             dis.close();
        }catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }
    //copy the file from HDFS to local
    public void GetFile(Configuration conf, String srcFile, String dstFile){
        try {
              FileSystem hdfs = FileSystem.get(conf);
              Path srcPath = new Path(srcFile);
              Path dstPath = new Path(dstFile);
              hdfs.copyToLocalFile(true,srcPath, dstPath);
        }catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }
    //copy the local file to HDFS
    public void PutFile(Configuration conf, String srcFile, String dstFile){
    try {
          FileSystem hdfs = FileSystem.get(conf);
          Path srcPath = new Path(srcFile);
          Path dstPath = new Path(dstFile);
          hdfs.copyFromLocalFile(srcPath, dstPath);
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }
    //create the new file
    public FSDataOutputStream CreateFile(Configuration conf, String FileName){
    try {    
          FileSystem hdfs = FileSystem.get(conf);
          Path path = new Path(FileName);
          FSDataOutputStream outputStream = hdfs.create(path);
          return outputStream;
        } catch (IOException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
        }
        return null;
    }
    //rename the file name
    public boolean ReNameFile(Configuration conf, String srcName, String dstName){
    try {
            Configuration config = new Configuration();
            FileSystem hdfs = FileSystem.get(config);
            Path fromPath = new Path(srcName);
            Path toPath = new Path(dstName);
            boolean isRenamed = hdfs.rename(fromPath, toPath);
            return isRenamed;
        }catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        return false;
    }
    //delete the file
    // tyep = true, delete the directory
    // type = false, delete the file
    public boolean DelFile(Configuration conf, String FileName, boolean type){
        try {
              FileSystem hdfs = FileSystem.get(conf);
              Path path = new Path(FileName);
              boolean isDeleted = hdfs.delete(path, type);
              return isDeleted;
        }catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        return false;
    }
    //Get HDFS file last modification time
    public long GetFileModTime(Configuration conf, String FileName){
    try{
              FileSystem hdfs = FileSystem.get(conf);
              Path path = new Path(FileName);
              FileStatus fileStatus = hdfs.getFileStatus(path);
              long modificationTime = fileStatus.getModificationTime();
              return modificationTime;
        }catch(IOException e){
            e.printStackTrace();
        }
        return 0;
    }
    //check if a file  exists in HDFS
    public boolean CheckFileExist(Configuration conf, String FileName){
    try{            
              FileSystem hdfs = FileSystem.get(conf);
              Path path = new Path(FileName);
              boolean isExists = hdfs.exists(path);
              return isExists;
        }catch(IOException e){
            e.printStackTrace();
        }
        return false;
    }
    //Get the locations of a file in the HDFS cluster
    public List<String []> GetFileBolckHost(Configuration conf, String FileName){
        try{
              List<String []> list = new ArrayList<String []>();
              FileSystem hdfs = FileSystem.get(conf);
              Path path = new Path(FileName);
              FileStatus fileStatus = hdfs.getFileStatus(path);
    
              BlockLocation[] blkLocations = hdfs.getFileBlockLocations(fileStatus, 0, fileStatus.getLen());
              
              int blkCount = blkLocations.length;
              for (int i=0; i < blkCount; i++) {
                String[] hosts = blkLocations[i].getHosts();
                list.add(hosts);
               }
              return list;
            }catch(IOException e){
                e.printStackTrace();
            }
            return null;
    }
    //Get a list of all the nodes host names in the HDFS cluster
    // have no authorization to do this operation
    public String[] GetAllNodeName(Configuration conf){
        try{
              FileSystem fs = FileSystem.get(conf);
              DistributedFileSystem hdfs = (DistributedFileSystem) fs;
              DatanodeInfo[] dataNodeStats = hdfs.getDataNodeStats();
              String[] names = new String[dataNodeStats.length];
              for (int i = 0; i < dataNodeStats.length; i++) {
                  names[i] = dataNodeStats[i].getHostName();
              }
              return names;
        }catch(IOException e){
            System.out.println("error!!!!");
            e.printStackTrace();
        }
        return null;
    }
}

package mapreduce;

import java.io.IOException;
import java.util.StringTokenizer;

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

import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class mywordcount {
    public static  class wordcountMapper extends
        Mapper<LongWritable, Text, Text, IntWritable>{
        private final static IntWritable one = new IntWritable(1);
        private Text word = new Text();
        public void map(LongWritable key, Text value, Context context)throws IOException, InterruptedException{
            String line = value.toString();
            StringTokenizer itr = new StringTokenizer(line);
            while(itr.hasMoreElements()){
                word.set(itr.nextToken());
                context.write(word, one);
            }
        }
    }
    public static  class wordcountReducer extends
        Reducer<Text, IntWritable, Text, IntWritable>{
        public void reduce(Text key, Iterable<IntWritable>values, Context context)throws IOException, InterruptedException{
            int sum = 0;
            for (IntWritable str : values){
                sum += str.get();
            }
            context.write(key, new IntWritable(sum));
        }
    }
    /**
     * 2 args, the file you want to count words from and the directory you want to save the result
     * @param args /home/hadooper/testmp/testtext /home/hadooper/testmp/testresult
     * @throws Exception
     */
    public static  void main(String args[])throws Exception{
        //首先定义两个临时文件夹，这里可以使用随机函数+文件名，这样重名的几率就很小。
        String dstFile = "temp_src";
        String srcFile = "temp_dst";
        //这里生成文件操作对象。
        HDFS_File file = new HDFS_File();
        
        Configuration conf = new Configuration();
        // must!!!  config the fs.default.name be the same to the value in core-site.xml
        conf.set("fs.default.name","hdfs://node1");
        conf.set("mapred.job.tracker","node1:54311");

        //从本地上传文件到HDFS,可以是文件也可以是目录
        file.PutFile(conf, args[0], dstFile);
        
        System.out.println("up ok");
        Job job = new Job(conf, "mywordcount");        
        job.setJarByClass(mywordcount.class);
        
        job.setInputFormatClass(TextInputFormat.class);
        
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        
        job.setMapperClass(wordcountMapper.class);
        job.setReducerClass(wordcountReducer.class);
        job.setCombinerClass(wordcountReducer.class);
        //注意这里的输入输出都应该是在HDFS下的文件或目录        
        FileInputFormat.setInputPaths(job, new Path(dstFile));
        FileOutputFormat.setOutputPath(job, new Path(srcFile));
//开始运行
        job.waitForCompletion(true);
        //从HDFS取回文件保存至本地
        file.GetFile(conf, srcFile, args[1]);
        System.out.println("down the result ok!");
//删除临时文件或目录
        file.DelFile(conf, dstFile, true);
        file.DelFile(conf, srcFile, true);
        System.out.println("delete file on hdfs ok!");
    }
}