一个wordcount程序轻松玩转MapReduce编程模型

青旅半醒 2022-06-04 04:07 230阅读 0赞

可以毫不夸张的说，几乎开发中绝大部分的MR程序都是基于wordcount编程模型而来，或者说用wordcount变化而来（改变的主要是业务方面的逻辑）。所以，熟练掌握wordcount编程模型，是掌握MR编程的基础。

**MAPREDUCE****示例编写及编程规范**

**1.1****编程规范**

**1用户编写的程序分成三个部分**：Mapper，Reducer，Driver(提交运行mr程序的客户端)

\- Mapper的输入数据是KV对的形式（KV的类型可自定义）

\- Mapper的输出数据是KV对的形式（KV的类型可自定义）

\- Mapper中的业务逻辑写在map()方法中

\- map()方法（maptask进程）对每一个<K,V>调用一次

\- Reducer的输入数据类型对应Mapper的输出数据类型，也是KV

\- Reducer的业务逻辑写在reduce()方法中

\- Reducetask进程对每一组相同k的<k,v>组调用一次reduce()方法

\- 用户自定义的Mapper和Reducer都要继承各自的父类

\- 整个程序需要一个Drvier来进行提交，提交的是一个描述了各种必要信息的job对象![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2NDQyNTUz_size_16_color_FFFFFF_t_70][]

**2.wordcount案例**

** 1.wcmapper**

package cn.itcast.bigdata.mr.wcdemo;

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Mapper;

**/\*\***

** \* KEYIN: 默认情况下，是mr框架所读到的一行文本的起始偏移量，Long,**

** \* 但是在hadoop中有自己的更精简的序列化接口，所以不直接用Long，而用LongWritable**

** \***

** \* VALUEIN:默认情况下，是mr框架所读到的一行文本的内容，String，同上，用Text**

** \***

** \* KEYOUT：是用户自定义逻辑处理完成之后输出数据中的key，在此处是单词，String，同上，用Text**

** \* VALUEOUT：是用户自定义逻辑处理完成之后输出数据中的value，在此处是单词次数，Integer，同上，用IntWritable**

** \* 因为hadoop分布式计算，所以数据要经过网络传输，所以要序列化，保证数据的一致性。**

\*/

public class WordcountMapper extends Mapper<LongWritable, Text, Text, IntWritable>\{

/\*\*

\* map阶段的业务逻辑就写在自定义的map()方法中

\***maptask会对每一行输入数据调用一次我们自定义的map()方法 。maptask将读到的一行一行的传送给map方法，实际上maptask是读了很多先缓存起来，在一行行发送给map方法**

\*/

@Override

protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException \{

**//将maptask传给我们的文本内容先转换成String**

String line = value.toString();

//根据空格将这一行切分成单词

String\[\] words = line.split(" ");

//将单词输出为<单词，1>

for(String word:words)\{

**//将单词作为key，将次数1作为value，以便于后续的数据分发，可以根据单词分发，以便于相同单词会到相同的reduce task**

context.write(new Text(word), new IntWritable(1));

** 2.wcreducer程序**

package cn.itcast.bigdata.mr.wcdemo;

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

/\*\*

** \* KEYIN, VALUEIN 对应  mapper输出的KEYOUT,VALUEOUT类型对应**

** \***

** \* KEYOUT, VALUEOUT 是自定义reduce逻辑处理结果的输出数据类型**

** \* KEYOUT是单词**

** \* VLAUEOUT是总次数**

\*/

public class WordcountReducer extends Reducer<Text, IntWritable, Text, IntWritable>\{

**  /\*\***

**     \* <angelababy,1><angelababy,1><angelababy,1><angelababy,1><angelababy,1>**

**     \* <hello,1><hello,1><hello,1><hello,1><hello,1><hello,1>**

**     \* <banana,1><banana,1><banana,1><banana,1><banana,1><banana,1>**

**     \* 入参key，是一组相同单词kv对的key ，一组一组的。同一单词放在迭代器里。**

**     \*/**

@Override

protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException \{

int count=0;

/\*Iterator<IntWritable> iterator = values.iterator();

while(iterator.hasNext())\{

count += iterator.next().get();  //序列化的对象的值不能直接获取，必须要用get（）方法

\}\*/

for(IntWritable value:values)\{

count += value.get();     //序列化的对象的值不能直接获取，必须要用get（）方法

context.write(key, new IntWritable(count));

**3.wcDriver**

package cn.itcast.bigdata.mr.wcdemo;

import org.apache.hadoop.conf.Configuration;

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;

/\*\*

\* 相当于一个yarn集群的客户端

\* 需要在此封装我们的mr程序的相关运行参数，指定jar包

\* 最后提交给yarn

\* @author

\*/

public class WordcountDriver \{ ** //不用输入参数，这种将输入输出路径写死的方式不推荐。**

public static void main(String\[\] args) throws Exception \{

if (args == null || args.length == 0) \{

args = new String\[2\];

args\[0\] = "hdfs://master:9000/wordcount/input/wordcount.txt";  //统计的文本在集群上

args\[1\] = "hdfs://master:9000/wordcount/output8";

Configuration conf = new Configuration();

//        conf.set("HADOOP\_USER\_NAME", "hadoop");

//        conf.set("dfs.permissions.enabled", "false");

/\*conf.set("mapreduce.framework.name", "yarn");

conf.set("yarn.resoucemanager.hostname", "mini1");\*/

Job job = Job.getInstance(conf);

/\*job.setJar("/home/hadoop/wc.jar");\*/

//指定本程序的jar包所在的本地路径

job.setJarByClass(WordcountDriver.class);

//指定本业务job要使用的mapper/Reducer业务类

job.setMapperClass(WordcountMapper.class);

job.setReducerClass(WordcountReducer.class);

//指定mapper输出数据的kv类型

job.setMapOutputKeyClass(Text.class);

job.setMapOutputValueClass(IntWritable.class);

//指定最终输出的数据的kv类型

job.setOutputKeyClass(Text.class);

job.setOutputValueClass(IntWritable.class);

//指定job的输入原始文件所在目录

FileInputFormat.setInputPaths(job, new Path(args\[0\]));

//指定job的输出结果所在目录

FileOutputFormat.setOutputPath(job, new Path(args\[1\]));

//将job中配置的相关参数，以及job所用的java类所在的jar包，提交给yarn去运行

**/\*job.submit();\*/不用写，因为job.waitfoCompletion(true)里面调用job.submit（）方法**

boolean res = job.waitForCompletion(true);

System.exit(res?0:1);

--------------------

/\*\*

\***相当于一个yarn集群的客户端，**

\* 需要在此封装我们的mr程序相关运行参数，指定jar包

\* 最后提交给yarn

\* @author Administrator

\*/

public class WordcountDriver**\{//运行需要输入参数**

public static void main(String\[\] args) throws Exception \{

//**1 获取配置信息，或者job对象实例**

**              /\*如果，不本地测试的话，直接集群运行，所有的关于configuration的都可不配置**

**                直接 Job job = new Job();即可，但实际开发中，把程序打包丢到集群上之前，要先**

**                测试一番，没有问题，才丢到集群上。**

**              \*/**

Configuration configuration = new Configuration();

Job job = Job.getInstance(configuration);

//**8 配置提交到yarn上运行,windows和Linux变量不一致，因为job在封装时会封装很多参数进取，而两者的参数类型不一样，不兼容，比如$在linux,在win中是%。**

**              ****//如果集群上运行程序不用配置下面关于yarn的代码，因为只要它提交的hadoop集群上，而每个结点已经配置下面这些文件，最终可以找到yarn.**

**              //如果是本地windows状态下运行程序，则不需要configuration配置置yarn所在节点。**

//            configuration.set("mapreduce.framework.name", "yarn");

//            configuration.set("yarn.resourcemanager.hostname", "hadoop103");

//**6 指定本程序的jar包所在的本地路径**

//            job.setJar("/home/robot/wc.jar");  与下面方法作用相同

** job.setJarByClass(WordcountDriver.class);**

//**2 指定本业务job要使用的mapper/Reducer业务类**

job.setMapperClass(WordcountMapper.class);

job.setReducerClass(WordcountReducer.class);

//**3 指定mapper输出数据的kv类型**

**             ****//注意map函数的输出类型默认情况下是和reduce函数的是相同的，所以如果mapper输出类型和reducer输入类型一样时，****可以不用设置setMapOutkey/value这一项。**

job.setMapOutputKeyClass(Text.class);

job.setMapOutputValueClass(IntWritable.class);

//**4 指定最终输出的数据的kv类型**

job.setOutputKeyClass(Text.class);

job.setOutputValueClass(IntWritable.class);

//**5 指定job的输入输出文件所在目录**

FileInputFormat.setInputPaths(job, new Path(args\[0\]));

** //上面等价于FileInputFormat.addInputPath(job,new Path(args\[0\]))  ,而且addInputFormat可以实现多路径的输入**

//这里的args\[\] 是 输入的参数位置。

FileOutputFormat.setOutputPath(job, new Path(args\[1\]));

//注意：这里输出路径在运行Job任务前是不能存在的，否则hadoop执行时会报错，并且拒绝执行job

//主要目的是为了防止数据丢失，因为长时间运行的作业如果结果被意外覆盖，那会是很悲催的。

** // 7 将job中配置的相关参数，以及job所用的java类所在的jar包， 提交给yarn去运行**

//   **job.submit();  **

**         //本地运行的话，则不用提交，会自动提交。****job.waitfoCompletion(true)里面调用job.submit（）**

System.exit( job.waitForCompletion(true)?0:1);

//退出系统，关闭连接，waitForCompletion()方法返回一个布尔值，表示执行的true或者false，这个布尔值被转换为程序的退出代码0或1.waitForCompletion()唯一的一个参数是一个标识，当标识为true时，作业会把其进度信息写到控制台。

--------------------

（4）将程序打成jar包，然后拷贝到hadoop集群中。

（5）启动hadoop集群

（6）执行wordcount程序

\[robot@hadoop102 software\]$ hadoop jar  wc.jar com.robot.wordcount.WordcountDriver /user/robot/input  /user/robot/output1

总结：注意在JobDriver中我们没有对输入的类型进行设置，可以通过输入格式来控制，因为默认使用的是TextInputForat(文本输入格式)

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2NDQyNTUz_size_16_color_FFFFFF_t_70]: /images/20220604/920795d9bfdd4cab857125a981fd4c36.png