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 HDFSpublic 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 localpublic 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 HDFSpublic 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 filepublic 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 namepublic 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 filepublic 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 timepublic 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 HDFSpublic 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 clusterpublic 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 operationpublic 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 extendsMapper<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 extendsReducer<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.xmlconf.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!");}
}