统计手机流量 - 使用bean传输

迷南。 2022-02-26 01:10 201阅读 0赞

### 统计手机流量 - 使用bean传输 ###

*  JavaBean
 *  Mapper
 *  reduce
 *  Main

> 上一篇博客中，使用的是键值对传输，但是，如果要字段很多怎么办呢？比如，需要上行流量、下行流量、总流量，这个时候就不能使用键值对的方式传输了，而是要将这些字段封装为一个JavaBean。

# JavaBean #

public class Traffic implements Writable{
            private String phone;
            private int upStream;
            private int downStream;
            private int total;
            public Traffic() {
            }
            public void setTraffic(String phone, int upStream, int downStream) {
                    this.phone = phone;
                    this.upStream = upStream;
                    this.downStream = downStream;
                    this.total = upStream + downStream;
            }
            public String getPhone() {
                    return phone;
            }
            public void setPhone(String phone) {
                    this.phone = phone;
            }
            public int getUpStream() {
                    return upStream;
            }
            public void setUpStream(int upStream) {
                    this.upStream = upStream;
            }
            public int getDownStream() {
                    return downStream;
            }
            public void setDownStream(int downStream) {
                    this.downStream = downStream;
            }
            public int getTotal() {
                    return total;
            }
            public void setTotal(int total) {
                    this.total = total;
            }
    
            @Override
            public void readFields(DataInput in) throws IOException {
                    upStream = in.readInt();
                    downStream = in.readInt();
                    total = in.readInt();
            }
    
            @Override
            public void write(DataOutput out) throws IOException {
                    out.writeInt(upStream);
                    out.writeInt(downStream);
                    out.writeInt(total);
            }
            @Override
            public String toString() {
                    return "[phone=" + phone + ", upStream=" + upStream + ", downStream=" + downStream + ", total=" + total + "]";
            }
    }

> java的序列化需要实现Serializable接口，但是hadoop提供了更加便于网路传输的Writable接口，实现这个接口要实现两个方法，一个是序列化，另一个是反序列化，写出和读入的字段顺序要完全一致。

# Mapper #

public class TrafficSumMapper extends Mapper<LongWritable, Text, Text, Traffic>{
    
            Traffic traffic = new Traffic();
            Text k = new Text();
    
            @Override
            protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
                    String line = value.toString();
                    String[] fileds = line.split("\t");
                    if(fileds.length!=11) return;
    
                    int upStream = Integer.parseInt(fileds[8]);
                    int downStream = Integer.parseInt(fileds[9]);
    
                    traffic.setTraffic(fileds[0], upStream, downStream);
                    k.set(fileds[0]);
                    context.write(k, traffic);
            }
    
    }

# reduce #

public class TrafficSumReduce extends Reducer<Text, Traffic, Text, Traffic>{
            @Override
            protected void reduce(Text key, Iterable<Traffic> values, Context context) throws IOException, InterruptedException {
                    Traffic t = new Traffic();
                    int totalUpStream = 0;
                    int totalDownStream = 0;
                    int total = 0;
                    for (Traffic traffic : values) {
                            totalUpStream += traffic.getUpStream();
                            totalDownStream += traffic.getDownStream();
                            total += traffic.getTotal();
                    }
                    t.setTraffic(key.toString(), totalUpStream, totalDownStream);
                    context.write(key, t);
            }
    }

# Main #

public class Main {
            public static void main(String[] args) throws Exception {
                    Configuration config = new Configuration();
    
                    Job job = Job.getInstance(config);
    
                    job.setJarByClass(TrafficCount.class);
    
                    job.setMapperClass(TrafficSumMapper.class);
                    job.setReducerClass(TrafficSumReduce.class);
    
                    job.setMapOutputKeyClass(Text.class);
                    job.setMapOutputValueClass(Traffic.class);
    
                    job.setOutputKeyClass(Text.class);
                    job.setOutputValueClass(Traffic.class);
    
                    FileInputFormat.setInputPaths(job, new Path("/home/hasee/bigdata/traffic/input/traffic.dat"));
                    FileOutputFormat.setOutputPath(job, new Path("/home/hasee/bigdata/traffic/output1"));
                    boolean res = job.waitForCompletion(true);
                    System.exit(res ? 0 : 1);
            }
    }