7月29日

今天突然发生问题一开始只有xshell连接hadoop102连接失败 103104是正常的我就点了几次那个xshell的打开然后vm就卡了开始未响应卡了很久我只能关机再开虚拟机还是连不上然后一下子又未响应又关机几次未响应之后三台都连不上了

经过好几个小时的调试，调试各种配置，我的所有配置都没有问题，，是虚拟机自己的卡顿，过了一段时间后就恢复了

后面学习了序列化，跟着学习了一个简单的案例

** map 和reduce在两个不同服务器上,传递bean数据需要进行序列化，将内存中的数据存储到硬盘，从另一个服务器上将硬盘数据反序列化到内存

要自己写一个类

FlowBean类
* 1 定义类实现writable接口
* 2 重写序列化和反序列化分方法
* 3 重写空参构造
* 4 toString方法
*/
public class FlowBean implements Writable {
    private long upFlow;//上行流量
    private long downFlow;//下行流量
    private long sumFlow;//总流量

    //空参构造
    public FlowBean() {
    }

    public long getUpFlow() {
        return upFlow;
    }

    public void setUpFlow(long upFlow) {
        this.upFlow = upFlow;
    }

    public long getDownFlow() {
        return downFlow;
    }

    public void setDownFlow(long downFlow) {
        this.downFlow = downFlow;
    }

    public long getSumFlow() {
        return sumFlow;
    }

    public void setSumFlow(long sumFlow) {
        this.sumFlow = sumFlow;
    }
    public void setSumFlow() {
        this.sumFlow = this.downFlow+this.upFlow;
    }
    @Override
    public void write(DataOutput out) throws IOException {
        out.writeLong(upFlow);
        out.writeLong(downFlow);
        out.writeLong(sumFlow);
    }

    @Override
    public void readFields(DataInput in) throws IOException {
        this.upFlow=in.readLong();
        this.downFlow=in.readLong();
        this.sumFlow=in.readLong();
    }

    @Override
    public String toString() {
        return upFlow +" " + downFlow + " " + sumFlow ;
    }
}

FlowMapper类

public class FlowMapper extends Mapper<LongWritable, Text,Text,FlowBean> {
    private Text outK=new Text();
    private FlowBean outv=new FlowBean();
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //1 获取一行
        String line=value.toString();
        //2 切割
        String[] split = line.split(" ");
        //3 抓取需要内容
        String phone=split[1];
        String up=split[split.length-3];
        String down=split[split.length-2];
        //4 封装
        outK.set(phone);
        outv.setUpFlow(Long.parseLong(up));
        outv.setDownFlow(Long.parseLong(down));
        outv.setSumFlow();
        //写出
        context.write(outK,outv);
    }
}

FlowReducer类

public class FlowReducer extends Reducer<Text,FlowBean,Text,FlowBean> {

    private FlowBean outv=new FlowBean();
    @Override
    protected void reduce(Text key, Iterable<FlowBean> values, Context context) throws IOException, InterruptedException {
        //1 遍历集合累加key相同量
        long allup=0;
        long alldown=0;
        for (FlowBean value : values) {
            allup+=value.getUpFlow();
            alldown+=value.getDownFlow();
        }
        //2 封装
        outv.setUpFlow(allup);
        outv.setDownFlow(alldown);
        outv.setSumFlow();

        //3 写出
        context.write(key,outv);
    }
}

FlowDriver类

public class FlowDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        //1 获取job对象
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);

        //2 设置jar包
        job.setJarByClass(FlowDriver.class);

        //3 关联mapper和reducer
        job.setMapperClass(FlowMapper.class);
        job.setReducerClass(FlowReducer.class);

        //4 设置mapper 输出的key和value的类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(FlowBean.class);

        //5 设置最终数据输出的key和value的类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(FlowBean.class);

        //6 设置数据的输入输出路径
        FileInputFormat.setInputPaths(job,new Path("E:\hadoop\Flow"));
        FileOutputFormat.setOutputPath(job,new Path("E:\hadoop\Flowout1"));
        //7 提交job
        boolean result = job.waitForCompletion(true);
        System.exit(result ? 0:1);
    }
}

学习时间：12：25到15：50