MapReduce单词统计 hadoop集群

package com.openyuan; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.util.GenericOptionsParser; import org.slf4j.Logger; import org.slf4j.LoggerFactory; /** * <p> * 主类 * <p> * * @author Open开袁 **/ public class WordMain { private static final Logger LOG = LoggerFactory.getLogger(WordMain.class); public static void main(String[] args) throws Exception { // 进行客户端身份的设置(root为虚拟机的用户名，hadoop集群节点的其中一个都可以) System.setProperty("HADOOP_USER_NAME","root"); // Configuration类：读取Hadoop的配置文件，如 site-core.xml...； // 也可用set方法重新设置（会覆盖）：conf.set("fs.default.name", "hdfs://xxxx:9000") Configuration conf = new Configuration(); // 将命令行中参数自动设置到变量conf中 String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs(); /* * 这里必须有输入输出 */ if (otherArgs.length != 2) { LOG.error("命令输入错误，请检查！"); System.exit(2); } // 删除输出目录 removeOutput(conf, otherArgs[1]); /* * 创建一个job */ Job job = new Job(conf, "word count"); // 新建一个 job，传入配置信息 job.setJarByClass(WordMain.class); // 设置 job 的主类 job.setMapperClass(WordMapper.class); // 设置 job 的 Mapper 类 job.setCombinerClass(WordCombiner.class); // 设置 job 的 作业合成类 job.setReducerClass(WordReducer.class); // 设置 job 的 Reducer 类 job.setOutputKeyClass(Text.class); // 设置 job 输出数据的关键类 job.setOutputValueClass(IntWritable.class); // 设置 job 输出值类 FileInputFormat.addInputPath(job, new Path(otherArgs[0])); // 文件输入 FileOutputFormat.setOutputPath(job, new Path(otherArgs[1])); // 文件输出 boolean result = false; try { result = job.waitForCompletion(true); } catch (Exception e) { LOG.error("", e); } LOG.info("{} 单词统计MapReduce {}", GlobalUtil.getFormatDate(), result ? "成功" : "失败"); System.exit(result ? 0 : 1); // 等待完成退出 } public static void removeOutput(Configuration conf, String outputPath) { try { // 通过FileSystem的静态get()方法获取HDFS文件系统客户端对象 FileSystem hdfs = FileSystem.get(conf); // 待删除目录路径(outputPath) Path src = new Path(outputPath); // 删除 hdfs.delete(src,true); } catch (Exception e) { throw new RuntimeException("删除异常"); } } }