mapreduce wordcount的java实现

淩亂°似流年 2023-06-20 09:17 176阅读 0赞

### mapreduce wordcount的java实现 ###

*   *   *   *  mapper方法实现如下：
             *  reducer方法实现如下：
             *  Driver方法实现如下：
             *  mapreduce的运行方法
             *  mapreduce的编程核心

整个实现过程可以用3个方法来实现：  
**mapper、reducer、driver**

#### mapper方法实现如下： ####

package com.aura.cn.mapreduce;
    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;
    /**
    *@class WordCountMapper.java
    *@author Samuel 
    *@version v1.0
    *@date 2019年12月9日下午4:03:39
    *@description   
    */
    	/**
    	 * 分而治之
    	 * 1）继承 Mapper 
    	 * 2)重写  map
    	 * 
    	 * 泛型：
    	 * 输入泛型   
    	 * 在进入到  mapper 之前 框架做了
    	 * 		mapreduce底层框架给的（流  读取）
    	 * 		line=br.readLine()
    	 * KEYIN,   输入的key的类型  每一行的标志  底层字节偏移量  每一行的起始偏移量  long
    	 * VALUEIN,   输入的value的类型  这里指的就是一行内容  string
    	 * 
    	 * 输出的泛型:  map端的处理结果  输出结果 给  Reducer
    	 * 	根据需求来的
    	 * KEYOUT,   输出的key的类型  这里指的是每一个单词  String
    	 * VALUEOUT  输出的value的类型  这里指的就是词频   int
    	 * 
    	 *注意：
    	 *整个mapreduce 中  使用hadoop中的类型 不能使用java中的类型
    	 *整个mapreduce 中  需要频繁的 数据持久化  网络传输的
    	 *要求 数据类型  具备 序列化  反序列化能力的
    	 *
    	 *java中  序列化  反序列化  重（工作量太大）
    	 *	Stu（zs,19）
    	 *hadoop中  提供了一个全新的序列化 、 反序列化 工具 ：  Writable 
    	 *	对于我们常用的一些类型  也已经帮实现了对应的序列化  反序列化的类型
    	 *	int ---IntWritable
    	 *	long---LongWritable
    	 *	double---DoubleWritable
    	 *	String---Text
    	 *	null---NullWritable
    	 *	......
    	 *序列化反序列化类型：java---Serializable
    	 *	1)固化（持久化）磁盘
    	 *	2）网络传输
    	 *
    	 *序列化： 对象--- 01010101
    	 *反序列化：0101---对象
    	 * 
    	 *  注意：整个mapreduce计算中 数据传输  都是 k-v键值对形式
    	 */
    public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable>{
        
    	/**
    	 * map端  核心逻辑
    	 * 
    	 * 1）参数
    	 * 参数1：一行偏移量
    	 * 参数2：一行内容
    	 * 参数3：上下文对象。  进行传输。  对上：框架 ，对下：reduce
    	 * 
    	 * 
    	 * 2）调用频率
    	 * 一行一次
    	 * 每次只能处理一行内容的
    	 * 
    	 * 
    	 * hello	word	hello	ww  key=0   value=hello	word	hello	ww
    		lily	hadoop
    		hadoop	spark
    		hive	spark
    		hive	hadoop
    	 */
    	@Override
    	protected void map(LongWritable key, Text value, Context context)
    			throws IOException, InterruptedException {
        
    		//实现逻辑
    		//获取一行内容 将hadoop类型--->java (Text--->String)
    		//br.readLine
    		String line = value.toString();
    		//切分每一个单词  一行单词
    		String[] words = line.split("\t");
    		//hello，1	word,1	hello,1	ww,1
    		//循环遍历每一个单词  封装k-v   发送给reduce
    		for(String w:words){
        
    			//write方法  发送的时候  k-v  参数1：发送的k 单词  参数2：发送的value 1
    			//将java--> hadoop
    			//String--Text
    			Text mk=new Text(w);
    			IntWritable mv=new IntWritable(1);
    			context.write(mk, mv);
    		}
    	}
    }

整个mapreduce计算中，数据的传输，都是以 k-v键值对形式进行  
参数很重要：

输入的泛型：
    * KEYIN,   输入的key的类型  每一行的标志  底层字节偏移量  每一行的起始偏移量  long
    * VALUEIN,   输入的value的类型  这里指的就是一行内容  string
    * 
    输出的泛型:  map端的处理结果  输出结果 给  Reducer，是根据需求来的。
    * KEYOUT,   输出的key的类型  这里指的是每一个单词  String
    * VALUEOUT  输出的value的类型  这里指的就是词频   int

#### reducer方法实现如下： ####

* 参数很重要：
     * key=hello  values==>  <1,1,1>
     * 参数1 key：每一组的一个key 
     * 参数2 values: 每一组中的  所有的value值
     * 参数3 context：上下文对象  上：map   下：输出结果  hdfs

#### Driver方法实现如下： ####

package com.aura.cn.mapreduce;
    import java.io.IOException;
    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;
    
    /**
    *@class Driver.java
    *@author Samuel 
    *@version v1.0
    *@date 2019年12月9日下午4:02:55
    *@description   
    */
    	/**
    	 * 组装map reduce 
    	 * 
    	 * 将  map   reduce 封装一个job中
    	 */
    public class Driver {
        
    	//args 接受 控制台传参   参数1 agrs[0]  参数2  args[1]  多个参数使用空格隔开
    	public static void main(String[] args) throws IllegalArgumentException, IOException, ClassNotFoundException, InterruptedException {
        
    		//获取集群的配置文件
    		Configuration conf=new Configuration();
    		
    		//启动job   构建一个job对象
    		Job job=Job.getInstance(conf);
    		
    		//进行job的封装  
    		//指定jar包运行的 主类
    		/*
    		 * 获取class 
    		 * 1)类名。class
    		 * 2）对象。getClass
    		 * 3)Class.forName()
    		 */
    		job.setJarByClass(Driver.class);
    		
    		//指定map 和 reduce对应的类
    		job.setMapperClass(WordCountMapper.class);
    		job.setReducerClass(WordCountReducer.class);
    		
    		//指定map输出的  k v的类型
    		/*
    		 * 框架读取文件 -----》 mapper ----》reducer
    		 * 泛型的作用周期：
    		 * 	编译时生效  运行时自动擦除
    		 */
    		job.setMapOutputKeyClass(Text.class);
    		job.setMapOutputValueClass(IntWritable.class);
    		
    		//指定reduce输出的类型（指定reduce最终输出的类型）
    		job.setOutputKeyClass(Text.class);
    		job.setOutputValueClass(IntWritable.class);
    		
    		//指定输入路径
    		FileInputFormat.addInputPath(job, new Path(args[0]));
    		
    		//指定输出路径
    		FileOutputFormat.setOutputPath(job, new Path(args[1]));
    		
    		//job.submit();//提交。也可以用此方法，但submit方法不会打印运行日志
    		
    		//提交。waitForCompletion方法的参数，代表 是否打印日志
    		job.waitForCompletion(true);
    	}
    }

Driver方法总结：

1、获取集群的配置文件
    2、构建一个job对象
    3、指定jar包运行的 主类；指定map 和 reduce对应的类
    4、指定reduce最终输出的类型
    5、指定输入、输出路径
    6、提交

#### mapreduce的运行方法 ####

1、打jar包  
2、将jar包上传到linux的一个节点  
3、运行  
语法：hadoop jar jar包名 运行主类名称（qualifided name）输入路径 输出路径  
eg:hadoop jar wc.jar com.aura.cn.mapreduce.Driver /input /output

#### mapreduce的编程核心 ####

1、map输出的key，决定reduce接收的数据格式。  
2、如何确定map的key值？  
按照什么进行分组，什么就是key  
例如一些关键词：每一个、相同等