案例二：基于MapReduce的分词统计
一、准备条件：
1、数据源：六个“.txt”诗经文件 2、Hadoop集群环境五、功能实现
前提准备：
引入类库将Hadoop包解压，并进入相对路径为hadoop-2.7.2\share\hadoop目录下，搜索*.jar,全选赋值到新建的lib目录下，并在该目录下搜索source.jar（为源码包），删除；并在该目录下搜索test（测试包），删除，剩下的就是可用的包。新建项目命名为Hadoop-mr-02 新建lib文件夹，将第一步所找到的jar包复制到该文件夹下，并引入到项目中在该项目中，将IKAnalyzer2012_u6.jar复制到该文件夹下，并引入到项目中第一步
新建TokenizerMapper.java类继承于Mapper<Object, Text, Text, IntWritable>
代码如下:
package com.xt.hadoop; import java.io.ByteArrayInputStream; import java.io.IOException; import java.io.InputStream; import java.io.InputStreamReader; import java.io.UnsupportedEncodingException; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Mapper.Context; import org.wltea.analyzer.core.IKSegmenter; import org.wltea.analyzer.core.Lexeme; public class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> { private final static IntWritable one = new IntWritable(1); private Text word = new Text(); private String pattern = "[^//w]"; // 正则表达式，代表不是0-9, a-z, // A-Z的所有其它字符,其中还有下划线 public void map(Object key, Text value, Context context) throws IOException, InterruptedException { String line = value.toString().toLowerCase(); // 全部转为小写字母 try { InputStream is = new ByteArrayInputStream(line.getBytes("UTF-8")); IKSegmenter seg = new IKSegmenter(new InputStreamReader(is), false); Lexeme lex = seg.next(); while (lex != null) { String text = lex.getLexemeText(); word.set(text); context.write(word, one); lex = seg.next(); } } catch (UnsupportedEncodingException e) { e.printStackTrace(); } catch (IOException e) { e.printStackTrace(); } } }第二步
新建IntSumReducer.java类继承于Reducer<Text, IntWritable, Text, IntWritable>
代码如下：
package com.xt.hadoop; import java.io.IOException; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; public class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> { private IntWritable result = new IntWritable(); public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { int sum = 0; for (IntWritable val : values) { sum += val.get(); } result.set(sum); context.write(key, result); } }第三步
新建HadoopWordCount.java类,实现main方法
具体代码如下：
package com.xt.hadoop; import java.io.IOException; 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; public class HadoopWordCount { public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { // TODO Auto-generated method stub // 作业 Job job = new Job(); job.setJarByClass(HadoopWordCount.class); // 设置作业名称 便于调试 job.setJobName("word count"); // 添加输入路径，可以添加多个路径 // 输入不进可以是具体文件，还可以是目录，不会递归 FileInputFormat.addInputPath(job, new Path(args[0])); // 设置输出路径，只能添加一个，而且不能存在 FileOutputFormat.setOutputPath(job, new Path(args[1])); // 设置mapper job.setMapperClass(TokenizerMapper.class); // 设置reducer job.setReducerClass(IntSumReducer.class); // 设置输出key类 job.setOutputKeyClass(Text.class); // 设置输出Value类 job.setOutputValueClass(IntWritable.class); // 等待作业完成 System.out.println(job.waitForCompletion(true)); } }六、项目部署
前提准备 1、将项目生成JAR文件，并将Jar文件放到master主机中 注：在生成中Jar包除了IKAnalyzer2012_u6.jar以外，其余全部不选择。 在第三步需要设置main函数类。 2、将诗经数据上传到hdfs文件系统中第一步：通过WinSCP将生成的Jar文件拷贝到/usr/local目录下 第二步：通过WinSCP将诗经数据文件靠背到/usr/local目录下
并执行命令： 1、在HDFS创建 `hadoop fs -mkdir -p /shijing` 2、将天气数据文件上传到HDFS /shijing目录下 hadoop fs -put shijing/* /shijing 3、执行Jar文件 hadoop jar hadoopmr02.jar /shijing /sout运行结果如下图所示
sout目录如下图所示
part-r-00000为数据文件 查看数据文件内如如下hadoop fs -cat /sout/part*
