Hadoop系列-MapReduce自定义数据类型（序列化、反序列化机制）（十二）

港控/mmm° 2022-06-13 05:24 236阅读 0赞

Github代码下载地址：

1，[JAVA工程代码][JAVA]

大家都知道，Hadoop中为Key的数据类型必须实现WritableComparable接口，而Value的数据类型只需要实现Writable接口即可；能做Key的一定可以做Value，能做Value的未必能做Key。但是具体应该怎么应用呢？本篇文章将结合手机上网流量业务进行分析。

先介绍一下业务场景：统计每个用户的上行流量和，下行流量和，以及总流量和。

本次描述所用数据，日志格式描述：手机号码，上行流量，下行流量

测试的具体数据如下：

![Center][]

接下来贴出详细代码，代码中含有详细注释，从代码中可以看出，用到了hadoop自定义的数据类型MyData，因为MyData只做value，所以在代码中只需要实现Writable接口。

package com.hadoop.minbo.mapreduce.custom2;
    
    import java.io.IOException;
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.FSDataInputStream;
    import org.apache.hadoop.fs.FileSystem;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.IOUtils;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
    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.output.FileOutputFormat;
    import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
    import org.apache.hadoop.mapreduce.lib.partition.HashPartitioner;
    
    /**
     * 方式一
     */
    public class FlowCount {
    
    	static class WordCountMapper extends Mapper<Object, Text, Text, MyData> {
    
    		@Override
    		protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {
    			// 拿到日志中的一行数据
    			String line = value.toString();
    			// 切分各个字段
    			String[] splited = line.split(" ");
    			// 获取我们所需要的字段:手机号、上行流量、下行流量
    			String msisdn = splited[0];
    			long upPayLoad = Long.parseLong(splited[1]);
    			long downPayLoad = Long.parseLong(splited[2]);
    			// 将数据进行输出
    			context.write(new Text(msisdn), new MyData(upPayLoad, downPayLoad));
    		}
    	}
    
    	static class WordCountReducer extends Reducer<Text, MyData, Text, MyData> {
    		@Override
    		protected void reduce(Text key, Iterable<MyData> values, Reducer<Text, MyData, Text, MyData>.Context context)
    				throws IOException, InterruptedException {
    			long payLoadSum = 0L; // 计算每个用户的上行流量和
    			long downLoadSum = 0L; // 统计每个用户的下行流量和
    			// 数据传递过来的时候：<手机号，{MyData1, MyData2, MyData3……}>
    			for (MyData md : values) {
    				payLoadSum += md.upPayLoad;
    				downLoadSum += md.downPayLoad;
    			}
    			// 在此需要重写toString()方法
    			context.write(key, new MyData(payLoadSum, downLoadSum)); 
    		}
    	}
    
    	public static String path1 = "input1";
    	public static String path2 = "output1";
    
    	public static void main(String[] args) throws Exception {
    		// Window下运行设置
    		System.setProperty("hadoop.home.dir", "F:\\hadoop\\hadoop-2.7.3"); // 设置hadoop安装路径
    		System.setProperty("HADOOP_USER_NAME", "hadoop"); // 用户名
    
    		Configuration conf = new Configuration();
    		FileSystem fileSystem = FileSystem.get(conf);
    		if (fileSystem.exists(new Path(path2))) {
    			fileSystem.delete(new Path(path2), true);
    		}
    
    		Job job = Job.getInstance(conf);
    		job.setJarByClass(FlowCount.class);
    
    		FileInputFormat.setInputPaths(job, new Path(path1));
    		job.setInputFormatClass(TextInputFormat.class);
    
    		job.setMapperClass(WordCountMapper.class);
    		job.setReducerClass(WordCountReducer.class);
    
    		//设置reduce执行任务数，默认1个
    		job.setNumReduceTasks(1);
    		//设置分区方式，默认hash
    		job.setPartitionerClass(HashPartitioner.class);
    
    		// 指定maptask的输出类型
    		job.setMapOutputKeyClass(Text.class);
    		job.setMapOutputValueClass(MyData.class);
    
    		// 指定reducetask的输出类型
    		job.setOutputKeyClass(Text.class);
    		job.setOutputValueClass(MyData.class);
    
    		job.setOutputFormatClass(TextOutputFormat.class);
    		FileOutputFormat.setOutputPath(job, new Path(path2));
    
    		job.waitForCompletion(true);
    
    		// 查看运行结果：
    		FSDataInputStream fr = fileSystem.open(new Path("output1/part-r-00000"));
    		IOUtils.copyBytes(fr, System.out, 2048, true);
    	}
    }

package com.hadoop.minbo.mapreduce.custom2;
    
    import java.io.DataInput;
    import java.io.DataOutput;
    import java.io.IOException;
    
    import org.apache.hadoop.io.Writable;
    
    /**
     * 自定义数据类型需要在hadoop中传输需要实现Writable接口
     * @author MINBO
     */
    public class MyData implements Writable {
    
    	public long upPayLoad; // 上行流量
    	public long downPayLoad; // 下行流量
    	public long loadSum; // 总流量
    
    	// 为了能够反序列化必须要定义一个无参数的构造函数
    	public MyData() {
    
    	}
    
    	public MyData(long upPayLoad, long downPayLoad) {
    		this.upPayLoad = upPayLoad;
    		this.downPayLoad = downPayLoad;
    		this.loadSum = upPayLoad + downPayLoad;// 利用构造函数的技巧，创建构造函数时，总流量被自动求出
    	}
    
    	// 只要数据在网络中进行传输，就需要序列化与反序列化
    	// 先序列化,将对象(字段)写到字节输出流当中
    	@Override
    	public void write(DataOutput fw) throws IOException {
    		fw.writeLong(upPayLoad);
    		fw.writeLong(downPayLoad);
    	}
    
    	// 反序列化,将对象从字节输入流当中读取出来，并且序列化与反序列化的字段顺序要相同
    	@Override
    	public void readFields(DataInput fr) throws IOException {
    		this.upPayLoad = fr.readLong();// 将上行流量给反序列化出来
    		this.downPayLoad = fr.readLong(); // 将下行流量给反序列化出来
    	}
    
    	@Override
    	public String toString() {
    		return "" + this.upPayLoad + "\t" + this.downPayLoad + "\t" + this.loadSum;
    	}
    }

查看运行结果：

![Center 1][]

从上面的实例可以看出，Hadoop中的自定义数据类型其实是很简单的，但是Hadoop为什么需要自己定义一套数据类型呢？原因在于：Java中的数据类型在序列化与反序列化的过程中太麻烦了。Java中的数据类型在序列化与反序列化的过程中必须要保证这些类与类之间的关系,从这个角度讲,意味着代码量就很大，数据在网络中传输就很占网宽，而hadoop认为这样太麻烦了,所以有自定义的数据类型,简化了序列化与反序列化的过程,保证了代码量的简洁。

其实如果对hadoop中的自定义数据类型不是很了解的话，我们也可以用现有的hadoop数据类型，比如收LongWritable，Text等来解决业务问题，比如对于上面给的手机上网流量统计业务，我们的代码也可以这么设计：

package com.hadoop.minbo.mapreduce.custom2;
    
    import java.io.IOException;
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.FSDataInputStream;
    import org.apache.hadoop.fs.FileSystem;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.IOUtils;
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
    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.output.FileOutputFormat;
    import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
    import org.apache.hadoop.mapreduce.lib.partition.HashPartitioner;
    
    /**
     * 方式二
     */
    public class FlowCount2 {
    
    	static class WordCountMapper extends Mapper<LongWritable, Text, Text, Text> {
    
    		@Override
    		protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context)
    				throws IOException, InterruptedException {
    			// 拿到日志中的一行数据
    			String line = value.toString();
    			// 切分各个字段
    			String[] splited = line.split(" ");
    			// 获取我们所需要的字段:手机号、上行流量、下行流量
    			String num = splited[0];
    			String upPayLoad = splited[1];
    			String downPayLoad = splited[2];
    			String str = "" + upPayLoad + " " + downPayLoad;// 这样改变即可
    			// 将数据进行输出
    			context.write(new Text(num), new Text(str));
    		}
    	}
    
    	static class WordCountReducer extends Reducer<Text, Text, Text, Text> {
    
    		@Override
    		protected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, Text>.Context context)
    				throws IOException, InterruptedException {
    			long payLoadSum = 0L; // 计算每个用户的上行流量和
    			long downLoadSum = 0L; // 统计每个用户的下行流量和
    			long sum = 0L;
    			for (Text v : values) {
    				String[] splited = v.toString().split(" ");
    				payLoadSum += Long.parseLong(splited[0]);
    				downLoadSum += Long.parseLong(splited[1]);
    			}
    
    			sum = payLoadSum + downLoadSum;
    			String result = "" + payLoadSum + " " + downLoadSum + " " + sum;
    			context.write(key, new Text(result));
    		}
    
    	}
    
    	public static String path1 = "input1";
    	public static String path2 = "output1";
    
    	public static void main(String[] args) throws Exception {
    		// Window下运行设置
    		System.setProperty("hadoop.home.dir", "F:\\hadoop\\hadoop-2.7.3"); // 设置hadoop安装路径
    		System.setProperty("HADOOP_USER_NAME", "hadoop"); // 用户名
    
    		Configuration conf = new Configuration();
    		FileSystem fileSystem = FileSystem.get(conf);
    		if (fileSystem.exists(new Path(path2))) {
    			fileSystem.delete(new Path(path2), true);
    		}
    
    		Job job = Job.getInstance(conf);
    		job.setJarByClass(FlowCount2.class);
    
    		FileInputFormat.setInputPaths(job, new Path(path1));
    		job.setInputFormatClass(TextInputFormat.class);
    
    		job.setMapperClass(WordCountMapper.class);
    		job.setReducerClass(WordCountReducer.class);
    
    		// 设置reduce执行任务数，默认1个
    		job.setNumReduceTasks(1);
    		// 设置分区方式，默认hash
    		job.setPartitionerClass(HashPartitioner.class);
    
    		// 指定maptask的输出类型
    		job.setMapOutputKeyClass(Text.class);
    		job.setMapOutputValueClass(Text.class);
    
    		// 指定reducetask的输出类型
    		job.setOutputKeyClass(Text.class);
    		job.setOutputValueClass(Text.class);
    
    		job.setOutputFormatClass(TextOutputFormat.class);
    		FileOutputFormat.setOutputPath(job, new Path(path2));
    
    		job.waitForCompletion(true);
    
    		// 查看运行结果：
    		FSDataInputStream fr = fileSystem.open(new Path("output1/part-r-00000"));
    		IOUtils.copyBytes(fr, System.out, 2048, true);
    	}
    }

运行结果：

![Center 2][]

从上面也说明了一个道理，对于知识的运用，是需要灵活的掌握。  
如果问题，欢迎指正！

参考资料：

[http://www.2cto.com/kf/201607/524055.html][http_www.2cto.com_kf_201607_524055.html]

[https://my.oschina.net/u/3294842/blog/858198][https_my.oschina.net_u_3294842_blog_858198]

[http://blog.csdn.net/huahai\_nb/article/details/55667732][http_blog.csdn.net_huahai_nb_article_details_55667732]

[http://www.cnblogs.com/ahu-lichang/p/6664875.html][http_www.cnblogs.com_ahu-lichang_p_6664875.html]

[JAVA]: https://github.com/hemin1003/hadoop-study/tree/master/src/com/hadoop/minbo/mapreduce/custom2
[Center]: /images/20220613/d15e70282a844f25a07db5c887bbaa52.png
[Center 1]: /images/20220613/3b7b3406f0d94cb3b49be0cdfbe0a2ea.png
[Center 2]: /images/20220613/c6da6199660544d8a6ed4bb24500a7d1.png
[http_www.2cto.com_kf_201607_524055.html]: http://www.2cto.com/kf/201607/524055.html
[https_my.oschina.net_u_3294842_blog_858198]: https://my.oschina.net/u/3294842/blog/858198
[http_blog.csdn.net_huahai_nb_article_details_55667732]: http://blog.csdn.net/huahai_nb/article/details/55667732
[http_www.cnblogs.com_ahu-lichang_p_6664875.html]: http://www.cnblogs.com/ahu-lichang/p/6664875.html