mapreduce程序案例1--flow.log计算上行流量和下行流量的排序

叁歲伎倆 2022-12-13 11:22 128阅读 0赞

1、flow.log计算上行流量和下行流量的排序

1363157985066 	13726230503	00-FD-07-A4-72-B8:CMCC	120.196.100.82	i02.c.aliimg.com		24	27	2481	24681	200
    1363157995052 	13826544101	5C-0E-8B-C7-F1-E0:CMCC	120.197.40.4			4	0	264	0	200
    1363157991076 	13926435656	20-10-7A-28-CC-0A:CMCC	120.196.100.99			2	4	132	1512	200
    1363154400022 	13926251106	5C-0E-8B-8B-B1-50:CMCC	120.197.40.4			4	0	240	0	200
    1363157993044 	18211575961	94-71-AC-CD-E6-18:CMCC-EASY	120.196.100.99	iface.qiyi.com	视频网站	15	12	1527	2106	200
    1363157995074 	84138413	5C-0E-8B-8C-E8-20:7DaysInn	120.197.40.4	122.72.52.12		20	16	4116	1432	200
    1363157993055 	13560439658	C4-17-FE-BA-DE-D9:CMCC	120.196.100.99			18	15	1116	954	200
    1363157995033 	15920133257	5C-0E-8B-C7-BA-20:CMCC	120.197.40.4	sug.so.360.cn	信息安全	20	20	3156	2936	200
    1363157983019 	13719199419	68-A1-B7-03-07-B1:CMCC-EASY	120.196.100.82			4	0	240	0	200
    1363157984041 	13660577991	5C-0E-8B-92-5C-20:CMCC-EASY	120.197.40.4	s19.cnzz.com	站点统计	24	9	6960	690	200
    1363157973098 	15013685858	5C-0E-8B-C7-F7-90:CMCC	120.197.40.4	rank.ie.sogou.com	搜索引擎	28	27	3659	3538	200
    1363157986029 	15989002119	E8-99-C4-4E-93-E0:CMCC-EASY	120.196.100.99	www.umeng.com	站点统计	3	3	1938	180	200
    1363157992093 	13560439658	C4-17-FE-BA-DE-D9:CMCC	120.196.100.99			15	9	918	4938	200
    1363157986041 	13480253104	5C-0E-8B-C7-FC-80:CMCC-EASY	120.197.40.4			3	3	180	180	200
    1363157984040 	13602846565	5C-0E-8B-8B-B6-00:CMCC	120.197.40.4	2052.flash2-http.qq.com	综合门户	15	12	1938	2910	200
    1363157995093 	13922314466	00-FD-07-A2-EC-BA:CMCC	120.196.100.82	img.qfc.cn		12	12	3008	3720	200
    1363157982040 	13502468823	5C-0A-5B-6A-0B-D4:CMCC-EASY	120.196.100.99	y0.ifengimg.com	综合门户	57	102	7335	110349	200
    1363157986072 	18320173382	84-25-DB-4F-10-1A:CMCC-EASY	120.196.100.99	input.shouji.sogou.com	搜索引擎	21	18	9531	2412	200
    1363157990043 	13925057413	00-1F-64-E1-E6-9A:CMCC	120.196.100.55	t3.baidu.com	搜索引擎	69	63	11058	48243	200
    1363157988072 	13760778710	00-FD-07-A4-7B-08:CMCC	120.196.100.82			2	2	120	120	200
    1363157985066 	13726238888	00-FD-07-A4-72-B8:CMCC	120.196.100.82	i02.c.aliimg.com		24	27	2481	24681	200
    1363157993055 	13560436666	C4-17-FE-BA-DE-D9:CMCC	120.196.100.99			18	15	1116	954	200

package application.mapreduce;
    
    import lombok.AllArgsConstructor;
    import lombok.Builder;
    import lombok.Data;
    import lombok.NoArgsConstructor;
    import org.apache.hadoop.io.WritableComparable;
    
    import java.io.DataInput;
    import java.io.DataOutput;
    import java.io.IOException;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2020, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2020-10-08 18:50
     * @Version: 1.0
     * @Description:
     */
    @Data
    @Builder
    @AllArgsConstructor
    @NoArgsConstructor
    public class Flow implements WritableComparable<Flow> {
        private String phone;
        private String mac;
        private String ip;
        private int upFlow;
        private int downFlow;
    
        @Override
        public int compareTo(Flow o) {
            return upFlow - o.upFlow;
        }
    
        @Override
        public void write(DataOutput dataOutput) throws IOException {
            dataOutput.writeUTF(phone);
            dataOutput.writeUTF(mac);
            dataOutput.writeUTF(ip);
            dataOutput.writeInt(upFlow);
            dataOutput.writeInt(downFlow);
        }
    
        @Override
        public void readFields(DataInput dataInput) throws IOException {
            phone = dataInput.readUTF();
            mac = dataInput.readUTF();
            ip = dataInput.readUTF();
            upFlow = dataInput.readInt();
            downFlow = dataInput.readInt();
        }
    }

package application.mapreduce;
    
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Mapper;
    
    import java.io.IOException;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2020, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2020-10-09 20:07
     * @Version: 1.0
     * @Description:
     */
    public class FlowMapper extends Mapper<LongWritable, Text,LongWritable, Flow> {
        private LongWritable longWritable = new LongWritable(0);
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = new String(value.getBytes(),"gb2312");
    
            line = line.replaceAll("[\t]+", " ");
    
            String[] fields = line.split(" ");
    
            String phone = fields[0];
            String mac = fields[2];
            String ip = fields[3];
            int upFlow = Integer.parseInt(fields[fields.length - 5]);
            int downFlow = Integer.parseInt(fields[fields.length - 4]);
    
            Flow flow = Flow.builder()
                    .phone(phone)
                    .mac(mac)
                    .ip(ip)
                    .upFlow(upFlow)
                    .downFlow(downFlow)
                    .build();
            longWritable.set(upFlow);
            context.write(longWritable,flow);
        }
    }

package application.mapreduce;
    
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.NullWritable;
    import org.apache.hadoop.io.Writable;
    import org.apache.hadoop.mapreduce.Reducer;
    
    import java.io.IOException;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2020, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2020-10-09 20:09
     * @Version: 1.0
     * @Description:
     */
    public class FlowReducer extends Reducer<LongWritable, Flow,NullWritable,Flow> {
        @Override
        protected void reduce(LongWritable key, Iterable<Flow> values, Context context) throws IOException, InterruptedException {
            for (Flow value : values){
                context.write(NullWritable.get(),value);
            }
        }
    }

package application.mapreduce;
    
    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.NullWritable;
    import org.apache.hadoop.mapreduce.Job;
    import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    
    import java.io.IOException;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2020, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2020-10-09 20:12
     * @Version: 1.0
     * @Description:
     */
    public class FlowDriver {
        public static void main(String[] args) throws Exception {
            Configuration configuration = new Configuration();
            Job job = Job.getInstance(configuration);
    
            job.setJarByClass(FlowDriver.class);
    
            job.setMapperClass(FlowMapper.class);
            job.setReducerClass(FlowReducer.class);
    
            job.setMapOutputKeyClass(LongWritable.class);
            job.setMapOutputValueClass(Flow.class);
    
           //job.setCombinerClass(FlowReducer.class);
    
            //job.setSortComparatorClass(FlowWritableComparator.class);
    
            job.setPartitionerClass(FlowPartitioner.class);
            job.setNumReduceTasks(3);
    
            job.setOutputKeyClass(NullWritable.class);
            job.setOutputValueClass(Flow.class);
    
            Path outputPath = new Path("D:/datas/output");
            FileSystem fileSystem = FileSystem.get(configuration);
            if (fileSystem.exists(outputPath)){
                fileSystem.delete(outputPath,true);
            }
    
            FileInputFormat.setInputPaths(job,new Path("D:/datas"));
            FileOutputFormat.setOutputPath(job,outputPath);
    
            boolean result = job.waitForCompletion(true);
            System.out.println(result);
        }
    }