Hadoop入门（八）自定义类型实例-统计手机流量数据Demo

矫情吗；* 2022-06-09 04:26 170阅读 0赞

业务需求：跟运营商后台数据，统计手机流量数据

下面是部分数据的截图：

![Center][]

我们把上面每一行数据抽象出来安“\\t” 隔开出11个字段：

![Center 1][]

我们要统计相同手机号的，6789字段的手机流量汇总，实际中其实数据非常庞大，Demo中我们就以上面4条为例，进行试验。

分析：K我们可以确定就是手机号，类型Text，V包括4个字段（6，7，8，9），那我们把V自定义成一个新类型。

我们自定义一个类：

package test;
    import java.io.DataInput;
    import java.io.DataOutput;
    import java.io.IOException;
    import java.net.URI;
    import java.net.URISyntaxException;
    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.io.Writable;
    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;
    
    public class TrafficSumApp {
    	//自定义类型（相当于java中的实体类）
    	private static class TrafficWritable implements Writable{
    		 //下面4个属性分别代表6,7,8,9字段
    		public long t1;
    		public long t2;
    		public long t3;
    		public long t4;		
    		public void write(DataOutput out) throws IOException {
    			out.writeLong(t1);
    			out.writeLong(t2);
    			out.writeLong(t3);
    			out.writeLong(t4);
    		}
    
    		public void set(long t12, long t22, long t32, long t42) {
    			this.t1 = t12;
    			this.t2 = t22;
    			this.t3 = t32;
    			this.t4 = t42;
    		}
    		public void set(String string1, String string2, String string3,
    				String string4) {
    			this.t1 = Long.parseLong(string1);
    			this.t2 = Long.parseLong(string2);
    			this.t3 = Long.parseLong(string3);
    			this.t4 = Long.parseLong(string4);
    		}
    		public void readFields(DataInput in) throws IOException {
    			this.t1 = in.readLong();
    			this.t2 = in.readLong();
    			this.t3 = in.readLong();
    			this.t4 = in.readLong();
    		}		
    		@Override
    		public String toString() {
    			return this.t1+"\t"+this.t2+"\t"+this.t3+"\t"+this.t4;
    		}
    	}
        //<k1,v1>
    	//<k2,v2>是手机号、单次通信的流量
    	//<k3,v3>是手机号、流量汇总
    	private static class TrafficMapper extends Mapper<LongWritable, Text, Text, TrafficWritable>{		
    		Text k2 = new Text(); 
    		TrafficWritable v2 = new TrafficWritable();//新建上面定义的类型对象
    		@Override
    		protected void map(
    				LongWritable key,
    				Text value,
    				Mapper<LongWritable, Text, Text, TrafficWritable>.Context context)
    				throws IOException, InterruptedException {
    			String line = value.toString();
    			String[] splited = line.split("\t");
    			//[1]手机号
    			k2.set(splited[1]);
    			//调用自定义类中的set方法把6789传入
    			v2.set(splited[6], splited[7], splited[8], splited[9]);
    			
    			context.write(k2, v2);
    		}
    	}
    	
    	private static class TrafficReducer extends Reducer<Text, TrafficWritable, Text, TrafficWritable>{
    		
    		TrafficWritable v3 = new TrafficWritable();
    		@Override
    		protected void reduce(
    				Text k2,
    				Iterable<TrafficWritable> v2s,
    				Reducer<Text, TrafficWritable, Text, TrafficWritable>.Context context)
    				throws IOException, InterruptedException {
    			
    			long t1 = 0L;
    			long t2 = 0L;
    			long t3 = 0L;
    			long t4 = 0L;
    			
    			for (TrafficWritable v2 : v2s) {
    				t1 += v2.t1;//累加
    				t2 += v2.t2;
    				t3 += v2.t3;
    				t4 += v2.t4;
    			}			
    			v3.set(t1, t2, t3, t4);			
    			context.write(k2, v3);
    		}
    	}
    	public static void deleteOutDir(Configuration conf, String OUT_DIR)
    			throws IOException, URISyntaxException {
    		FileSystem fileSystem = FileSystem.get(new URI(OUT_DIR), conf);
    		if(fileSystem.exists(new Path(OUT_DIR))){
    			fileSystem.delete(new Path(OUT_DIR), true);
    		}
    	}
    
    	public static void main(String[] args) throws Exception {
    			Configuration conf = new Configuration();
    			Job job = Job.getInstance(conf, TrafficSumApp.class.getSimpleName());
    			job.setJarByClass(TrafficSumApp.class);
    			//源文件存放位置
    			FileInputFormat.setInputPaths(job, new Path("hdfs://192.168.19.128:9000/telephone.dat"));
    			job.setInputFormatClass(TextInputFormat.class);
    			job.setMapperClass(TrafficMapper.class);
    			job.setMapOutputKeyClass(Text.class);
    			job.setMapOutputValueClass(TrafficWritable.class);
    			
    			job.setReducerClass(TrafficReducer.class);
    			job.setOutputKeyClass(Text.class);
    			job.setOutputValueClass(TrafficWritable.class);
    			//结果数据存放位置
    			String OUT_DIR = "hdfs://192.168.19.128:9000/outtel.dat";
    			FileOutputFormat.setOutputPath(job, new Path(OUT_DIR));
    			job.setOutputFormatClass(TextOutputFormat.class);		
    			deleteOutDir(conf, OUT_DIR);		
    			job.waitForCompletion(true);
    		}
    
    }

最后，打成jar,在Hadoop环境下运行即可。详细操作在我其他文章里有写，连接http://blog.csdn.net/zhangyunfeixyz/article/details/77151083

[Center]: /images/20220609/3533246ca47049a4b5d014668490ab42.png
[Center 1]: /images/20220609/4b7436122be647128b85285fb4ceb22d.png