MapReduce-day01

1. MapReduce的思想

　　MapReduce的思想就是“分而治之”，他适用于大量复杂的任务场景（大规模数据的处理场景）。MapReduce是一款分布式运算框架，核心功能是将用户编写的业务代码和自带的默认组件整合成一个完整的分布式计算程序，并发在hadoop集群上。

• Map负责“分”，就是把复杂的任务分成若干个简单的任务，来并行处理，前提是这些小任务可以并行计算，彼此间几乎没有依赖关系。
• Reduce负责“合”，就是把map阶段的结果进行全局汇总。

2.MapReduce的“天龙八部”

　　map阶段2步

• 第一部：设置InputFormat的类型（通常为TextInputformat）和数据的输入路径--获取数据的过程（得到k1，v1）
• 第二部：自定义Mapper--将k1，v1转换为k2，v2

shuffle阶段4步

• 第三部：分区的动作，如果有多个reduce才需要分区，默认只有一个。
• 第四部：排序，默认对K2进行排序（字典排序）
• 第五部：规约，combine是一个局部的reduce，map端的合并，是对mapreduce的优化操作，前提是不会影响结果，他可以减少网络传输。
• 第六部：分组：相同的K（k2）对应的value会放在同一个集合中--map传递的k2，v2变成新的k2，v2

Reduce阶段2步

• 第七部：自定义Reducer得到k2，v2将k2，v2转为k3，v3
• 第八部：设置OutPutFormat和数据的路径--生成结果文件

3. 入门案例单词计数

package com;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;



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.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;


import java.io.IOException;


public class WorldCount {
    static class WordCountMapper extends Mapper<LongWritable, Text,Text,LongWritable> {
        Text k=new Text();
        LongWritable v=new LongWritable();
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            String[] lines = line.split(" ");
            for (String s : lines) {
                k.set(s);
                v.set(1);
                context.write(k,v);
            }
        }
    }

    static class WordCountReducer extends Reducer<Text,LongWritable,Text,LongWritable>{
        Text k=new Text();
        LongWritable v=new LongWritable();

        @Override
        protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {
            long count=0;
            for (LongWritable value : values) {
                count+=value.get();
            }
            k.set(key);
            v.set(count);
            context.write(k,v);
        }
    }

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration conf = new Configuration();
        //创建任务
        Job job = Job.getInstance(conf, "wordCount");
        //设置输入类型路径
        job.setInputFormatClass(TextInputFormat.class);
        TextInputFormat.addInputPath(job,new Path("D:\hadoop_data\wc"));

        //设置map输出类型
        job.setMapperClass(WordCountMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(LongWritable.class);

        //设置reduce和输出类型
        job.setReducerClass(WordCountReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(LongWritable.class);

        //设置输出路径
        job.setOutputFormatClass(TextOutputFormat.class);
        TextOutputFormat.setOutputPath(job,new Path("D:\hadoop_output\wc"));

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

4. 深入MapReduce

a. 输入输出

MapReduce框架运行在<Key,Value>上，输入的是一对<Key,Value>，输出也是一对<Key,Value>，这两组键值对不一定相同。

• 输入一般按行读取key：行偏移量，value：行内容
• 在map阶段根据业务需求key->key2，value->value2，可能会有combiner操作，默认是没有的。
• reduce阶段将key2->key3，value2->value3

b. 执行流程

1. 按照规则进行逻辑切片，默认切片大小和block块大小一致，每个切片由一个maptask处理，所以一定要控制好切片数量
2. 把切片解析成k，v的形式，默认是把一行文本内容解析成键值对。
3. 每个k，v键值对都会调用一次map方法，多次调用会产生多个键值对
4. 按照一定规则对第三阶段的键值对进行分区，默认只有一个分区，分区的数量就是ReduceTask的数量
5. 对分区中的键值对排序，首先按照键排序，键相同再按照值排序。
6. 对数据进行局部聚合处理，也就是combiner，键相等的会调用一次reduce方法，相同键会被合并，这样数据量就会减少，默认本阶段是没有的。
7. Reduce任务会主动复制Mapper任务的输出。
8. 把复制过来的数据进行合并，在对合并后的数据进行排序。
9. 对排序后的数据调用reduce方法，键相等的数据会调用一次reduce方法。最后把这些文件写到hdfs中。

4. MapReduce的序列化

序列化：把结构化对象转换为字节流对象。

反序列化：把字节流对象转换为结构化对象。

注意：序列化与反序列化时顺序应该一样。

public void readFields(DataInput in) throws IOException { 
   
  upflow = in.readLong(); 
  dflow = in.readLong(); 
  sumflow = in.readLong(); 
   
 } 
 /**   * 序列化的方法 
  */ 
 @Override 
 public void write(DataOutput out) throws IOException { 
 
  out.writeLong(upflow); 
  out.writeLong(dflow); 
  out.writeLong(sumflow); 
 } 

5. 分区和排序

package cn.itcast.mr.flow;

import org.apache.hadoop.io.WritableComparable;

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

/**
 * @ClassName FlowBean
 * @Description 手机流量的POJO 序列化和排序
 */
public class FlowBean implements WritableComparable<FlowBean>{
    private Long upFlow;//上行总流量
    private Long downFlow;//下行总流量
    private Long sumFlow;//总流量

    public FlowBean() {
    }

    public FlowBean(Long upFlow, Long downFlow) {
        this.upFlow = upFlow;
        this.downFlow = downFlow;
        this.sumFlow = upFlow + downFlow;
    }

    /**
     * @Description 排序  pojo  pojo    默认是升序排序
     * @param int 如果返回值是0，代表两个数据相等，如果返回1 代表前面比后面的大，如果返回-1，代表前面比后面小
     * @return
     **/
    @Override
    public int compareTo(FlowBean o) {
        //如果两个值相等返回的结果就是0 如果前面比后面大，需要调整位置（得到的是正数），如果后面的大，得到的就是负数，就不需要调整
        return (int)(this.sumFlow.longValue()-o.sumFlow.longValue());
    }

    /**
     * @Description  序列化的方法
     * @param
     * @return
     **/
    @Override
    public void write(DataOutput out) throws IOException {
        out.writeLong(upFlow);
        out.writeLong(downFlow);
        out.writeLong(sumFlow);
    }

    /**
     * @Description  反序列化和序列化的顺序要保持一致
     * @param
     * @return
     **/
    @Override
    public void readFields(DataInput in) throws IOException {
        this.upFlow = in.readLong();
        this.downFlow = in.readLong();
        this.sumFlow = in.readLong();
    }


    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;
    }

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

package cn.itcast.mr.flow.partition;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;

/**
 * @ClassName MyPartitionner
 * @Description TODO
 */
public class MyPartitionner extends Partitioner<Text,Text> {
    @Override
    public int getPartition(Text text, Text text2, int i) {
        //text手机号，以起始3位数进行分区划分
        if(text.toString().startsWith("134")){
            return 0;
        }else if(text.toString().startsWith("135")){
            return 1;
        }else if(text.toString().startsWith("136")){
            return 2;
        }else if(text.toString().startsWith("137")){
            return 3;
        }else if(text.toString().startsWith("138")){
            return 4;
        }else {
            return 5;
        }
    }
}

package cn.itcast.mr.flow.partition;

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

import java.io.IOException;

/**
 * @ClassName PartitionMapper
 * @Description TODO
 *
 * K1       V1
 * 0        13719199419    240    0    240
 * -------------------------------
 * K2               V2
 * 13719199419      13719199419    240    0    240
 */
public class PartitionMapper extends Mapper<LongWritable,Text,Text,Text> {
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String[] splits = value.toString().split("	");
        context.write(new Text(splits[0]),value);
    }
}

package cn.itcast.mr.flow.partition;

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

import java.io.IOException;

/**
 * @ClassName PartitionReducer
 * @Description TODO
 *
 * K2               V2
 * 13719199419      13719199419    240    0    240
 * -----------------------------
 * K3                           V3
 * 13719199419    240    0    240     null
 *
 */
public class PartitionReducer extends Reducer<Text,Text,Text,NullWritable> {
    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        //将V2变成K3
        for (Text value : values) {
            context.write(value,NullWritable.get());
        }
    }
}

package cn.itcast.mr.flow.partition;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;

/**
 * @ClassName PartitionRunner
 * @Description TODO
 */
public class PartitionRunner {
    public static void main(String[] args) throws IOException, URISyntaxException, ClassNotFoundException, InterruptedException {
        Configuration conf = new Configuration();
        //创建任务
        Job job = Job.getInstance(conf, "partition");
        job.setJarByClass(PartitionRunner.class);//在集群上运行要加上
        //设置的分区个数少于自定义的获取值，抛异常、如果正好没问题，如果大于，会出现空白文件
        job.setNumReduceTasks(6);

        //第一步：设置InputFormat和数据的输入路径
        job.setInputFormatClass(TextInputFormat.class);
        TextInputFormat.addInputPath(job,new Path("hdfs://node01:9000/hadoop32/part-r-00000"));
        //第二步：设置自定义的Mapper类并且设置输出类型
        job.setMapperClass(PartitionMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        //第三步 设置分区
        job.setPartitionerClass(MyPartitionner.class);
        // 第四、五、六步 省略


        //第七步：设置自定义的Reducer类并且设置输出类型
        job.setReducerClass(PartitionReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);

        //第八步：设置OutputFormat和数据的输出路径
        job.setOutputFormatClass(TextOutputFormat.class);
        //强调，MR默认输出路径是不能存在的，如果存在会报错
        TextOutputFormat.setOutputPath(job,new Path("hdfs://node01:9000/hadoop32/partition"));
        //得到该文件夹
        FileSystem fileSystem = FileSystem.get(new URI("hdfs://node01:9000"), conf);
        if(fileSystem.exists(new Path("hdfs://node01:9000/hadoop32/partition"))){
            fileSystem.delete(new Path("hdfs://node01:9000/hadoop32/partition"), true);
        }

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