Hadoop概要——MapReduce&Yarn（二）

1. Java序列化的目的是为了实现对象的永久性存储与网络传输
2. 序列化主要是对对象的属性进行序列化
3. 序列化的本质是IO流，内存中的对象落盘形成文件对应的是输出流，磁盘中的文件形成内存中的对象对应的是输入流；ObjectOutputstream writeObject(Object obj)序列化方法，ObjectInputStream readObject()反序列化方法
4. Java序列化是一个重量级序列化，结果包含了（属性值，校验和，继承体系，头信息），数据比较大

自定义JavaBean的必要条件：①实现Serializable接口，②属性私有化，③空参构造器（序列化及反序列化需要），④get/set方法

Hadoop中的序列化是一种轻量级的序列化，只包含属性值和校验和

1. 紧凑：高效使用存储空间
2. 快速：读写数据的额外开销小
3. 互操作：支持多语言的交互

Hadoop中的JavaBean序列化需要满足以下条件：

1. 实现Writable接口，若同时需要自定义key的排序规则需要实现WritableComparable接口
2. 属性私有化
3. 空参构造器（序列化及反序列化需要）
4. get/set方法
5. 重写write(DataOutput out)和readFields(DataInput in)方法，注意：序列化的顺序需要与反序列化的顺序保持一致
6. 如果需要将自定义的JavaBean落盘，需要重写toString()方法
7. 如果需要将自定义的JavaBean作为key进行传输，由于shuffle需要实现WritableComparable接口，并且重写compareTo()方法

样例需求：统计每一个手机号耗费的总上行流量、总下行流量以及总流量

//输入数据:第一列为序列号，第二列为手机号码，第三列为IP地址，第四列为网址，第五列为上传流量，第六列为下载流量，第七列为状态码
1	13736230513	192.196.100.1	www.atguigu.com	2481	24681	200
2	13846544121	192.196.100.2			264	0	200
3 	13956435636	192.196.100.3			132	1512	200
4 	13966251146	192.168.100.1			240	0	404
5 	18271575951	192.168.100.2	www.atguigu.com	1527	2106	200
6 	84188413	192.168.100.3	www.atguigu.com	4116	1432	200
7 	13590439668	192.168.100.4			1116	954	200
8 	15910133277	192.168.100.5	www.hao123.com	3156	2936	200
9 	13729199489	192.168.100.6			240	0	200
10 	13630577991	192.168.100.7	www.shouhu.com	6960	690	200
11 	15043685818	192.168.100.8	www.baidu.com	3659	3538	200
12 	15959002129	192.168.100.9	www.atguigu.com	1938	180	500
13 	13560439638	192.168.100.10			918	4938	200
14 	13470253144	192.168.100.11			180	180	200
15 	13682846555	192.168.100.12	www.qq.com	1938	2910	200
16 	13992314666	192.168.100.13	www.gaga.com	3008	3720	200
17 	13509468723	192.168.100.14	www.qinghua.com	7335	110349	404
18 	18390173782	192.168.100.15	www.sogou.com	9531	2412	200
19 	13975057813	192.168.100.16	www.baidu.com	11058	48243	200
20 	13768778790	192.168.100.17			120	120	200
21 	13568436656	192.168.100.18	www.alibaba.com	2481	24681	200
22 	13568436656	192.168.100.19			1116	954	200

//JavaBean
import org.apache.hadoop.io.Writable;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

public class PhoneBean implements Writable {
    private Integer id;
    private String phoneNum;
    private String ip;

    private String website;
    private Double upload;
    private Double download;
    private Double total;
    private String status;

    public PhoneBean() {
        id = 0;
        phoneNum = "";
        ip = "";
        website = "";
        upload = 0.0;
        download = 0.0;
        total = 0.0;
        status = "";
    }

    public Integer getId() {
        return id;
    }

    public void setId(Integer id) {
        this.id = id;
    }

    public String getPhoneNum() {
        return phoneNum;
    }

    public void setPhoneNum(String phoneNum) {
        this.phoneNum = phoneNum;
    }

    public String getIp() {
        return ip;
    }

    public void setIp(String ip) {
        this.ip = ip;
    }

    public String getWebsite() {
        return website;
    }

    public void setWebsite(String website) {
        this.website = website;
    }

    public Double getUpload() {
        return upload;
    }

    public void setUpload(Double upload) {
        this.upload = upload;
    }

    public Double getDownload() {
        return download;
    }

    public void setDownload(Double download) {
        this.download = download;
    }

    public Double getTotal() {
        return total;
    }

    public void setTotal(Double total) {
        this.total = total;
    }

    public void setTotal() {
        this.total = this.upload + this.download;
    }

    public String getStatus() {
        return status;
    }

    public void setStatus(String status) {
        this.status = status;
    }

    @Override
    public void write(DataOutput out) throws IOException {
        out.writeInt(id);
        out.writeUTF(phoneNum);
        out.writeUTF(ip);
        out.writeUTF(website);
        out.writeDouble(upload);
        out.writeDouble(download);
        out.writeDouble(total);
        out.writeUTF(status);
    }

    @Override
    public void readFields(DataInput in) throws IOException {
        id = in.readInt();
        phoneNum = in.readUTF();
        ip = in.readUTF();
        website = in.readUTF();
        upload = in.readDouble();
        download = in.readDouble();
        total = in.readDouble();
        status = in.readUTF();
    }

    @Override
    public String toString() {
        return phoneNum + "t" + upload + "t" + download + "t" + total;
    }
}

//Mapper
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

public class PhoneMapper extends Mapper {
    PhoneBean outValue;
    private Text outKey;

    @Override
    protected void setup(Mapper.Context context) throws IOException, InterruptedException {
        outValue = new PhoneBean();
        outKey = new Text();

    }

    @Override
    protected void map(LongWritable key, Text value, Mapper.Context context) throws IOException, InterruptedException {
        String s = value.toString();
        String[] split = s.split("t");
        outKey.set(split[1]);
        outValue.setDownload(Double.parseDouble(split[split.length - 2]));
        outValue.setUpload(Double.parseDouble(split[split.length - 3]));
        context.write(outKey, outValue);
    }
}

//Reducer
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class PhoneReducer extends Reducer {
    PhoneBean outValue;

    @Override
    protected void setup(Reducer.Context context) throws IOException, InterruptedException {
        outValue = new PhoneBean();
    }

    @Override
    protected void reduce(Text key, Iterable values, Reducer.Context context) throws IOException, InterruptedException {
        Double totalUp = 0.0;
        Double totalDown = 0.0;
        for (PhoneBean value : values) {
            totalDown += value.getDownload();
            totalUp += value.getUpload();
        }
        outValue.setUpload(totalUp);
        outValue.setDownload(totalDown);
        outValue.setTotal();

        context.write(key, outValue);
    }
}

//Driver
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
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;

import java.io.IOException;

public class Driver {
    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        Configuration configuration = new Configuration();
        Job job = Job.getInstance(configuration);

        job.setJarByClass(Driver.class);
        job.setMapperClass(PhoneMapper.class);
        job.setReducerClass(PhoneReducer.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(PhoneBean.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(PhoneBean.class);

        FileInputFormat.setInputPaths(job, new Path("E:\DevProjects\Idea_Project\尚硅谷大数据\Hadoop\PhoneJavaBean\src\main\java\Phone"));
        FileOutputFormat.setOutputPath(job, new Path("E:\DevProjects\Idea_Project\尚硅谷大数据\Hadoop\PhoneJavaBean\src\main\java\output"));

        boolean result = job.waitForCompletion(true);
        System.exit(result ? 0 : 1);
    }
}

//结果
13470253144		180.0	180.0	360.0
13509468723		7335.0	110349.0	117684.0
13560439638		918.0	4938.0	5856.0
13568436656		3597.0	25635.0	29232.0
13590439668		1116.0	954.0	2070.0
13630577991		6960.0	690.0	7650.0
13682846555		1938.0	2910.0	4848.0
13729199489		240.0	0.0	240.0
13736230513		2481.0	24681.0	27162.0
13768778790		120.0	120.0	240.0
13846544121		264.0	0.0	264.0
13956435636		132.0	1512.0	1644.0
13966251146		240.0	0.0	240.0
13975057813		11058.0	48243.0	59301.0
13992314666		3008.0	3720.0	6728.0
15043685818		3659.0	3538.0	7197.0
15910133277		3156.0	2936.0	6092.0
15959002129		1938.0	180.0	2118.0
18271575951		1527.0	2106.0	3633.0
18390173782		9531.0	2412.0	11943.0
84188413		4116.0	1432.0	5548.0