MapReduce的计数器详解大数据

//计数器使用~解决：判断下输入文件中有多少hello 
Counter counterHello = context.getCounter("Sensitive words","hello"); 
//假设hello为敏感词 
            if(line != null && line.contains("hello")){ 
                counterHello.increment(1L); 
            } 
//计数器代码结束

  1 package Mapreduce; 
  2  
  3 import java.io.IOException; 
  4  
  5 import org.apache.hadoop.conf.Configuration; 
  6 import org.apache.hadoop.fs.Path; 
  7 import org.apache.hadoop.io.LongWritable; 
  8 import org.apache.hadoop.io.Text; 
  9 import org.apache.hadoop.mapreduce.Counter; 
 10 import org.apache.hadoop.mapreduce.Job; 
 11 import org.apache.hadoop.mapreduce.Mapper; 
 12 import org.apache.hadoop.mapreduce.Reducer; 
 13 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; 
 14 import org.apache.hadoop.mapreduce.lib.input.TextInputFormat; 
 15 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 
 16 import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; 
 17  
 18 /** 
 19  * 
 20  * 计数器的使用及测试 
 21  */ 
 22 public class MyCounterTest { 
 23     public static void main(String[] args) throws Exception { 
 24         //必须要传递的是自定的mapper和reducer的类，输入输出的路径必须指定，输出的类型<k3,v3>必须指定 
 25         //2将自定义的MyMapper和MyReducer组装在一起 
 26         Configuration conf=new Configuration(); 
 27         String jobName=MyCounterTest.class.getSimpleName(); 
 28         //1首先寫job，知道需要conf和jobname在去創建即可 
 29         Job job = Job.getInstance(conf, jobName); 
 30          
 31         //*13最后，如果要打包运行改程序，则需要调用如下行 
 32         job.setJarByClass(MyCounterTest.class); 
 33          
 34         //3读取HDFS內容：FileInputFormat在mapreduce.lib包下 
 35         FileInputFormat.setInputPaths(job, new Path("hdfs://neusoft-master:9000/data/hellodemo")); 
 36         //4指定解析<k1,v1>的类（谁来解析键值对） 
 37         //*指定解析的类可以省略不写，因为设置解析类默认的就是TextInputFormat.class 
 38         job.setInputFormatClass(TextInputFormat.class); 
 39         //5指定自定义mapper类 
 40         job.setMapperClass(MyMapper.class); 
 41         //6指定map输出的key2的类型和value2的类型  <k2,v2> 
 42         //*下面两步可以省略，当<k3,v3>和<k2,v2>类型一致的时候,<k2,v2>类型可以不指定 
 43         job.setMapOutputKeyClass(Text.class); 
 44         job.setMapOutputValueClass(LongWritable.class); 
 45         //7分区(默认1个)，排序，分组，规约 采用 默认 
 46          
 47         //接下来采用reduce步骤 
 48         //8指定自定义的reduce类 
 49         job.setReducerClass(MyReducer.class); 
 50         //9指定输出的<k3,v3>类型 
 51         job.setOutputKeyClass(Text.class); 
 52         job.setOutputValueClass(LongWritable.class); 
 53         //10指定输出<K3,V3>的类 
 54         //*下面这一步可以省 
 55         job.setOutputFormatClass(TextOutputFormat.class); 
 56         //11指定输出路径 
 57         FileOutputFormat.setOutputPath(job, new Path("hdfs://neusoft-master:9000/out3")); 
 58          
 59         //12写的mapreduce程序要交给resource manager运行 
 60         job.waitForCompletion(true); 
 61     } 
 62     private static class MyMapper extends Mapper<LongWritable, Text, Text,LongWritable>{ 
 63         Text k2 = new Text(); 
 64         LongWritable v2 = new LongWritable(); 
 65         @Override 
 66         protected void map(LongWritable key, Text value,//三个参数 
 67                 Mapper<LongWritable, Text, Text, LongWritable>.Context context)  
 68                 throws IOException, InterruptedException { 
 69             String line = value.toString(); 
 70             //计数器使用~解决：判断下输入文件中有多少hello  这里仅仅是举例，如果有很多的hello可能显示的还是如此结果 
 71             Counter counterHello = context.getCounter("Sensitive words","hello");//假设hello为敏感词 
 72             if(line != null && line.contains("hello")){ 
 73                 counterHello.increment(1L); 
 74             } 
 75             //计数器代码结束 
 76             String[] splited = line.split("/t");//因为split方法属于string字符的方法，首先应该转化为string类型在使用 
 77             for (String word : splited) { 
 78                 //word表示每一行中每个单词 
 79                 //对K2和V2赋值 
 80                 k2.set(word); 
 81                 v2.set(1L); 
 82                 context.write(k2, v2); 
 83             } 
 84         } 
 85     } 
 86     private static class MyReducer extends Reducer<Text, LongWritable, Text, LongWritable> { 
 87         LongWritable v3 = new LongWritable(); 
 88         @Override //k2表示单词，v2s表示不同单词出现的次数，需要对v2s进行迭代 
 89         protected void reduce(Text k2, Iterable<LongWritable> v2s,  //三个参数 
 90                 Reducer<Text, LongWritable, Text, LongWritable>.Context context) 
 91                 throws IOException, InterruptedException { 
 92             long sum =0; 
 93             for (LongWritable v2 : v2s) { 
 94                 //LongWritable本身是hadoop类型，sum是java类型 
 95                 //首先将LongWritable转化为字符串，利用get方法 
 96                 sum+=v2.get(); 
 97             } 
 98             v3.set(sum); 
 99             //将k2,v3写出去 
100             context.write(k2, v3); 
101         } 
102     } 
103 }