Hadoop之——自定义计数器

转载请注明出处:http://blog.csdn.net/l1028386804/article/details/46057909

1、Mapper类的实现

 /*** KEYIN  即k1     表示行的偏移量* VALUEIN    即v1     表示行文本内容* KEYOUT 即k2     表示行中出现的单词* VALUEOUT 即v2     表示行中出现的单词的次数，固定值1*/static class MyMapper extends Mapper<LongWritable, Text, Text, LongWritable>{protected void map(LongWritable k1, Text v1, Context context) throws java.io.IOException ,InterruptedException {Counter counter = context.getCounter("Sensitive word", "hello");String line = v1.toString();if(line.contains("hello")){//记录敏感词汇出现在一行中counter.increment(1);}final String[] splited = line.split("\t");for (String word : splited) {context.write(new Text(word), new LongWritable(1));}};}

2、Reducer类的实现

 /*** KEYIN  即k2     表示行中出现的单词* VALUEIN  即v2     表示行中出现的单词的次数* KEYOUT    即k3     表示文本中出现的不同单词* VALUEOUT  即v3     表示文本中出现的不同单词的总次数**/static class MyReducer extends Reducer<Text, LongWritable, Text, LongWritable>{protected void reduce(Text k2, java.lang.Iterable<LongWritable> v2s, Context ctx) throws java.io.IOException ,InterruptedException {long times = 0L;for (LongWritable count : v2s) {times += count.get();}ctx.write(k2, new LongWritable(times));};}

3、程序入口Main方法

 //程序入口Mainfanpublic static void main(String[] args) throws Exception {Configuration conf = new Configuration();final FileSystem fileSystem = FileSystem.get(new URI(INPUT_PATH), conf);final Path outPath = new Path(OUT_PATH);//如果已经存在输出文件，则先删除已存在的输出文件if(fileSystem.exists(outPath)){fileSystem.delete(outPath, true);}final Job job = new Job(conf , WordCount.class.getSimpleName());//1.1指定读取的文件位于哪里FileInputFormat.setInputPaths(job, INPUT_PATH);//指定如何对输入文件进行格式化，把输入文件每一行解析成键值对job.setInputFormatClass(TextInputFormat.class);//1.2 指定自定义的map类job.setMapperClass(MyMapper.class);//map输出的<k,v>类型。如果<k3,v3>的类型与<k2,v2>类型一致，下面两行代码可以省略job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(LongWritable.class);//1.3 分区job.setPartitionerClass(HashPartitioner.class);//有一个reduce任务运行job.setNumReduceTasks(1);//1.4 TODO 排序、分组//1.5 TODO 规约//2.2 指定自定义reduce类job.setReducerClass(MyReducer.class);//指定reduce的输出类型job.setOutputKeyClass(Text.class);job.setOutputValueClass(LongWritable.class);//2.3 指定写出到哪里FileOutputFormat.setOutputPath(job, outPath);//指定输出文件的格式化类job.setOutputFormatClass(TextOutputFormat.class);//把job提交给JobTracker运行job.waitForCompletion(true);}

4、完整程序

package com.lyz.hadoop.count;import java.net.URI;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Counter;
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;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.mapreduce.lib.partition.HashPartitioner;
/*** 利用Hadoop MapReduce统计文本中每个单词的数量* 自定义计数器* @author liuyazhuang*/
public class WordCount{//要统计的文件位置static final String INPUT_PATH = "hdfs://liuyazhuang:9000/d1/hello";//统计结果输出的位置static final String OUT_PATH = "hdfs://liuyazhuang:9000/out";//程序入口Mainfanpublic static void main(String[] args) throws Exception {Configuration conf = new Configuration();final FileSystem fileSystem = FileSystem.get(new URI(INPUT_PATH), conf);final Path outPath = new Path(OUT_PATH);//如果已经存在输出文件，则先删除已存在的输出文件if(fileSystem.exists(outPath)){fileSystem.delete(outPath, true);}final Job job = new Job(conf , WordCount.class.getSimpleName());//1.1指定读取的文件位于哪里FileInputFormat.setInputPaths(job, INPUT_PATH);//指定如何对输入文件进行格式化，把输入文件每一行解析成键值对job.setInputFormatClass(TextInputFormat.class);//1.2 指定自定义的map类job.setMapperClass(MyMapper.class);//map输出的<k,v>类型。如果<k3,v3>的类型与<k2,v2>类型一致，下面两行代码可以省略job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(LongWritable.class);//1.3 分区job.setPartitionerClass(HashPartitioner.class);//有一个reduce任务运行job.setNumReduceTasks(1);//1.4 TODO 排序、分组//1.5 TODO 规约//2.2 指定自定义reduce类job.setReducerClass(MyReducer.class);//指定reduce的输出类型job.setOutputKeyClass(Text.class);job.setOutputValueClass(LongWritable.class);//2.3 指定写出到哪里FileOutputFormat.setOutputPath(job, outPath);//指定输出文件的格式化类job.setOutputFormatClass(TextOutputFormat.class);//把job提交给JobTracker运行job.waitForCompletion(true);}/*** KEYIN  即k1     表示行的偏移量* VALUEIN    即v1     表示行文本内容* KEYOUT 即k2     表示行中出现的单词* VALUEOUT 即v2     表示行中出现的单词的次数，固定值1*/static class MyMapper extends Mapper<LongWritable, Text, Text, LongWritable>{protected void map(LongWritable k1, Text v1, Context context) throws java.io.IOException ,InterruptedException {Counter counter = context.getCounter("Sensitive word", "hello");String line = v1.toString();if(line.contains("hello")){//记录敏感词汇出现在一行中counter.increment(1);}final String[] splited = line.split("\t");for (String word : splited) {context.write(new Text(word), new LongWritable(1));}};}/*** KEYIN 即k2     表示行中出现的单词* VALUEIN  即v2     表示行中出现的单词的次数* KEYOUT    即k3     表示文本中出现的不同单词* VALUEOUT  即v3     表示文本中出现的不同单词的总次数**/static class MyReducer extends Reducer<Text, LongWritable, Text, LongWritable>{protected void reduce(Text k2, java.lang.Iterable<LongWritable> v2s, Context ctx) throws java.io.IOException ,InterruptedException {long times = 0L;for (LongWritable count : v2s) {times += count.get();}ctx.write(k2, new LongWritable(times));};}}

5、核心

Hadoop之——自定义计数器相关推荐

Hadoop学习笔记—7.计数器与自定义计数器
一.Hadoop中的计数器计数器:计数器是用来记录job的执行进度和状态的.它的作用可以理解为日志.我们通常可以在程序的某个位置插入计数器,用来记录数据或者进度的变化情况,它比日志更便利进行分析. ...
Hadoop之自定义数据类型
2019独角兽企业重金招聘Python工程师标准>>> 1.写一个类实现Writable接口 2.重写write和readFilelds方法 3.自定义数据类型,提供相应的gette ...
MapReduce 自定义计数器
MapReduce 允许用户编写程序来定义计数器,计数器的值可在 mapper 或 reduce 中增加,计数器由一个 Java 枚举(enum)类型来定义,以便对有关的计数器分组,一个作业可以定义的 ...
MapReduce的自定义计数器
1. 在map端使用计数器进行统计: import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; impor ...
Hadoop之Join、计数器、数据清洗概述
Hadoop之Join.计数器.数据清洗概述目录 Reduce join Map join 计数器应用数据清洗(ETL) 1. Reduce join 原理 Map端的主要工作:为来自不同表(文件 ...
Hadoop学习笔记—8.Combiner与自定义Combiner
一.Combiner的出现背景 1.1 回顾Map阶段五大步骤在第四篇博文<初识MapReduce>中,我们认识了MapReduce的八大步凑,其中在Map阶段总共五个步骤,如下图所示: ...
Hadoop MapReduce编程 API入门系列之压缩和计数器（三十）
不多说,直接上代码. Hadoop MapReduce编程 API入门系列之小文件合并(二十九) 生成的结果,作为输入源. 代码 package zhouls.bigdata.myMapReduce. ...
Hadoop教程（四）：理解MapReduce、MapReduce计数器和连接、MapReduce Hadoop程序连接数据
本教程中的代码分为 3 个部分: 解释 SalesMapper 类解释 SalesCountryReducer 类解释 SalesCountryDriver 类 SalesMapper类的说明在 ...
Vue自定义InputNumber 计数器组件
1.为什么要自己封装一个InputNumber 计数器组件? 因为原始的el-element的el-input-number组件有问题: 原生组件能输入英文,不能限制只能输入数值: 原始组件能通过键盘 ...

Hadoop之——自定义计数器

1、Mapper类的实现

2、Reducer类的实现

3、程序入口Main方法

4、完整程序

5、核心

Hadoop之——自定义计数器相关推荐

最新文章

热门文章