public abstract class FileInputFormat<K, V> extends InputFormat<K, V> 
public class TextInputFormat extends FileInputFormat<LongWritable, Text> 

Rich leaning form  --每条记录对应的键值对---> （0，Rich leaning form ）

@Override
public RecordReader<LongWritable, Text> createRecordReader(InputSplit split,TaskAttemptContext context) {
  String delimiter = context.getConfiguration().get("textinputformat.record.delimiter");
  byte[] recordDelimiterBytes = null;
  if (null != delimiter)
    recordDelimiterBytes = delimiter.getBytes(Charsets.UTF_8);
  return new LineRecordReader(recordDelimiterBytes);
}

View Code

return Math.max(minSize, Math.min(maxSize, blockSize));

CombineTextInputFormat 改变了切片规则；虽重写了从切片到k,v但返回的还是<LongWritable, Text>；适用于小文件过多的场景；

View Code

KeyValueTextInputFormat 改变了切片到k,v值；每一行均为一条记录，被分隔符分割为key，value；

NLineInputFormat 按行切片；切片-->k,v用默认方式；（不再按块去划分，而是按NLineInputFormat指定的行数N来划分即：输入文件的总行数 / N = 切片数）

一个文件一个切片，RecordReader把切片变成k,v值；
默认输入输出类型：InputFormat从数据源头的类型就改变了；

View Code

2、 MapReduce工作流程

在有限的内存实现全排序；

Job.split、wc.jar、Job.xml等文件通过客户端提交给yarn，yarn根据切片信息启动相应数量的MapTask；
MapTask1处理切片128M的数据块，它调用默认的TextInputFormat的RecoderReader类型的方法读取数据，它一行行读到的 k,v值（行号，行内容）给Mapper（）；

Mapper()调用map(k,v)方法 通过context.write(k, v)写出去（给框架） ----通过outputCollector收集（MapTask的处理结果）<K，V，P>到环形缓冲区中；

排序的时候要先完成分区（就是把分区号一样的在一块）；再在分区内部排序-->分区排序--局部的快排序（内存保证输出的每个文件内部是有序的，按照文件key做全排序，相同的key挨一起），
缓冲区满就溢出到文件磁盘（文件特点：①一个文件；②分区且区内有序；溢写完之后文件分两个部分spill.index索引（文件有多少个分区，分区从哪结束）和spill.out输出文件）

---->>多次溢出（取决于map方法）形成的多个文件，不同文件之间是无序的--->多个溢写的文件合并成一个文件，按照分区归并排序（A、B文件之间做比较，若A第一个值 < B第一个值，就写到C中；接着再比较下面的依次...）

首先分区1进行归并写到文件的前半部分，分区2再进行归并写到文件的后半部分...（归并是数据流）
到此为止，Map阶段结束（生成一个分区且区内有序的完整的输出文件，shuffle的前半部分）；

MapTask输出文件分区且区内有序，多个MapTask形成多个分区且区内有序的文件；多个map输出文件交给Reduce来处理，全都下载到本地；

--->>根据分区数启动相应数量的ReduceTask，ReduceTask1下载所有分区patition0即分区1的数据拷贝到内存缓冲区buffer，内存不够溢出到磁盘；ReduceTask2下载所有分区2的数据；
---->>各自合并文件，归并排序（多个文件归并得到一个文件，文件特点是：按照key有序）；
至此完成了从map输出到reduce输入的3次排序--全排序；有了这个全排序文件才能进行分组-->输出到reduce()方法里边；这就是shuffle的全过程；

--->>一次读取一组，按照相同key分组 --> GroupingComparator(k, knext) 分组 --->>Reduce(k,v) context.write(k,v)-->> 
经过outputformat（默认TextOutputFormat）RecordWriter--- write(k, v) --->>Part-r-00000文件;

如果启动了combiner：
第一次排序结束后在落盘（溢出写入磁盘）之前会经过combiner的合并（没有重复的单词，但两个ReduceTask之间有重复的单词）；
第二次排序即归并排序完之后还没落盘之前（没有重复的的单词）数据流会流入combiner（之后可进行压缩），写到磁盘上；

第三次排序，两个相同分区的数据下载（有可能有重复单词）合并---> 启动Reduce，没有combine

shuffle是一部分MapTask，一部分ReduceTask；MapTask会有一个单独的输出文件；ReduceTask要从MapTask中下载数据到本地，要等到所有的MapTask运行完之后才能运行；
按key分组，全排序---->按key排好，两个数据就可以决定分组，(默认调用compareTo进行分组，否则就用自定义的；)这个数据跟下个数据比较，一样就分同一组，不一样就新启动一组；
内存里边只有两个数据，这一个和下一个；

从map到reduce进行的全排序；快排，单轴排序，java用的双轴排序；
分区只是决定这条数据去哪个reduceTask，不能决定reduceTask的数量；
怎么比较key是否相等；

public class HashPartitioner<K, V> extends Partitioner<K, V> {
  public int getPartition(K key, V value,  int numReduceTasks) {
    return (key.hashCode() & Integer.MAX_VALUE) %  numReduceTasks;
  }
　　默认分区是根据 Key的hashCode对ReduceTask个数取模得到，没法控制哪个key存储到哪个分区；

}

 return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;    
numReduceTasks是设置的ReduceTask的数量；取余
Integer.MAX_VALUE换算成位是一串011111...；假设key.hashCode是
0100100，它们做位运算，没变化，但如果是以1开头的，就是负数，取余还是负数；  --->目的就是防止出现负数；

public class MyPartitioner extends Partitioner<Text, FlowBean> {
    //自定义分区
    public int getPartition(Text text, FlowBean flowBean, int i) {
        String substring = text.toString().substring(0, 3);
        switch (substring){
            case "136":
                return 0;
            case "137":
                return 1;
            case "138":
                return 2;
            case "139":
                return 3;
            default:
                return 4;
        }
    }
}


public class PatitionDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        //1.实例化
        Job job = Job.getInstance(new Configuration());
        //2.设置类jar路径
        job.setJarByClass(PatitionDriver.class);
        //3.设置Mapper和Reduce路径
        job.setMapperClass(FlowMapper.class);
        job.setReducerClass(FlowReduce.class);

        job.setPartitionerClass(MyPartitioner.class);
        //job.setNumReduceTasks(5);
        //job.setNumReduceTasks(6); //ReduceTask数量6 > 分区数量5; 产生空的输出文件part-r-00005
        //job.setNumReduceTasks(4); //ReduceTask数量4 < 分区数量5;  有部分数据无处放Illegal partition for 18271575951 (4)
                                    //如果分区数量跳过3，不按顺序；Illegal partition for 13956435636 (4);分区号不能跳，会浪费；分区号必须从0开始，逐一累加;
        job.setNumReduceTasks(1);   //ReduceTask数量1，不管MapTask输出多少个分区文件，最终结果都交给这一个ReduceTask，最终也只会产生一个part-r-0000文件
        //4.输出类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(FlowBean.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(FlowBean.class);
        //5.路径
        FileInputFormat.setInputPaths(job, new Path("F://input"));
        FileOutputFormat.setOutputPath(job, new Path("F://output"));
        //6.提交
        boolean b = job.waitForCompletion(true);
        System.exit(b ? 0 : 1);
    }
}

排序接口 WritableComparable，利用了Shuffle强制排序的过程

分区-->排序；把需要排序的放到key的位置就会自动排序（强制性）

/OrderBeam是WritableComparable的实现类
//bean对象做为key传输，需要实现WritableComparable接口重写compareTo方法，就可以实现排序。
　　public class OrderBean implements WritableComparable<OrderBean> { }
 
@Override
    public int compareTo(OrderBean o) {
        //先按orderById分; 如果相同则按price
        int compare = this.getOrderById().compareTo(o.getOrderById());
        if (compare == 0){
            return -Double.compare(this.price, o.price); //默认是升序; 降序需加-
        }else {
            return compare;
        }
    }
 @Override  //序列化把它写进去
    public void write(DataOutput dataOutput) throws IOException {
        dataOutput.writeUTF(orderById);
        dataOutput.writeUTF(productId);
        dataOutput.writeDouble(price);

    }
 @Override //反序列化要再把它读出来
    public void readFields(DataInput dataInput) throws IOException {
        orderById = dataInput.readUTF();
        productId = dataInput.readUTF();
        price = dataInput.readDouble();
    }

排序的同时可自定义分区
public class SortPatitioner extends Partitioner<FlowBean, Text>{
    @Override
    public int getPartition(FlowBean flowBean, Text text, int i) {
        String start = text.toString().substring(0, 3);
        switch (start) {
            case "136":
                return 0;
            case "137":
                return 1;
            case "138":
                return 2;
            case "139":
                return 3;
            default:
                return 4;
        }
    }
}

aaa　　aaa　　aaa

bbb　　bbb　　bbb

ccc　　ccc　　ccc
在map阶段是（aaa，1）3；（bbb, 1）3；（ccc, 1)3
　　　　用了combine在map阶段（aaa, 3)，（bbb, 3），（ccc, 3)

job.setCombinerClass(WcReduce.class);  //不设置CombinerClass

④分组（辅助排序）

分组规则和排序规则一致就不需要自定义分组；排序规则和分组规则不一样，就用到自定义分组；

分组粒度 > 排序的粒度;分组的粒度更粗一点，排序的粒度更细点；

public class MyComparator extends WritableComparator{
    public MyComparator(){
        super(OrderBean.class, true);
    }
    @Override
    public int compare(WritableComparable a, WritableComparable b) {
        OrderBean oa = (OrderBean) a;
        OrderBean ob = (OrderBean) b;
        return oa.getOrderById().compareTo(ob.getOrderById());  //分组

    }
}

public class OrderReducer extends Reducer<OrderBean, NullWritable, OrderBean, NullWritable>{

    @Override
    protected void reduce(OrderBean key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {

// context.write(key, NullWritable.get()); //key之前是相同的，不相同的key进入同一个组

        for (NullWritable value : values) {
            context.write(key, value);
        }
对于reduce，key, value都只有一个对象；
它拿到的数据是已经排好序的k，v对象；往下遍历一个，就反序列化一个，
再遍历一个，继续往下移动（只能往下移动，不能返回），反序列化； 对象始终只有两个，对象的值却是在不断变化（通过反序列化的方法）；
执行hasNaxt，调用WritableComparable跟下面比较下：若相等返回true就继续往下迭代；把当前的迭代值与下一个比较，0返回true继续迭代，非0返回false迭代结束；
结束后第二个reduceTask接着迭代的数据往下...分组和读数据是同时进行的；

Iterator<NullWritable> iterator = values.iterator();  //单例模式，来回遍历还是他自己
        for (int i = 0; i < 2; i++){
            if (!iterator.hasNext())
                break;
                iterator.next();
                context.write(key, NullWritable.get());  //查看前两个值
//取前两个值；
        }

NullWritalbe单例模式；私有化构造器，公有化一个get方法；饿汉式模式；

在大数据背景下，不允许有大量对象存在内存，在整个MapReduce框架运行过程中，所有的数据都是用序列化来传递的。需要用到对象时，现场序列化现场反序列化；
对于自定义的comparator比较的时候并没有现成的对象用，需要先构造两个空对象，拿到序列化的数据，反序列化进行比较；核心：减小IO；

public class MyOutputFormat extends FileOutputFormat<LongWritable, Text> {
    @Override
    public RecordWriter<LongWritable, Text> getRecordWriter(TaskAttemptContext job) throws IOException, InterruptedException {
        MyRecordWriter recordWriter = new MyRecordWriter();
        recordWriter.initialize(job);
        return recordWriter;
    }
}


public class MyRecordWriter extends RecordWriter<LongWritable, Text> {
    private FSDataOutputStream atguigu;
    private FSDataOutputStream others;
    public void initialize(TaskAttemptContext job) throws IOException {  //job来获取configuration
        //获取文件系统
        Configuration configuration = job.getConfiguration();
        FileSystem fileSystem = FileSystem.get(configuration);
        //获取文件目录
        String path = configuration.get(FileOutputFormat.OUTDIR);
   
        atguigu = fileSystem.create(new Path(path + "/atguigu.log"));
        others = fileSystem.create(new Path(path + "/others.log"));
    }
    @Override
    public void write(LongWritable key, Text value) throws IOException, InterruptedException {
        String line = value.toString() + "\n";
        if (line.contains("atguigu")){
            //包含就写出去; FSDataOutputStream要以bytes的格式写出去; 把k, v键值对变成text；
            atguigu.write(line.getBytes());
        }else {
            others.write(line.getBytes());
        }
    }
    @Override
    public void close(TaskAttemptContext context) throws IOException, InterruptedException {
        IOUtils.closeStream(atguigu);
        IOUtils.closeStream(others);
    }
}


public class OutDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Job job = Job.getInstance(new Configuration());
        job.setJarByClass(OutDriver.class);
        job.setOutputFormatClass(MyOutputFormat.class);
        FileInputFormat.setInputPaths(job, new Path("F://input"));
        FileOutputFormat.setOutputPath(job, new Path("F://output"));
        boolean b = job.waitForCompletion(true);
        System.exit(b ? 0 :1);
    }
}

 @Override
    public int compareTo(TableBean o) {
        //先按pid排序，再按名字排
        int compare = this.pid.compareTo(o.pid);
        if (compare == 0){
            return -this.pname.compareTo(o.pname);  //默认是升序，没名字的在前；降序加- ；把有pname的靠前；；
        }else {
            return compare;
        }
    }

在map阶段完成封装，首先按pid排序，然后再按pname排；
01　　小米        排完序之后（排序和分组）应该这样子； 把排好序的一块输到reduce里；
1001　　01　　1
1004　　02　　4


public class TableReducer extends Reducer<TableBean, NullWritable, TableBean, NullWritable> {
    @Override
    protected void reduce(TableBean key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
        Iterator<NullWritable> iterator = values.iterator();
        iterator.next();  //先遍历第一个，指针首先是指向0的，next完之后它就指向了1，
        String pname = key.getPname(); //取出pname;把剩下的数据写出去，因为它们都要进行替换
        System.out.println(pname);
        while (iterator.hasNext()){
            iterator.next(); //先执行下；
            key.setPname(pname);   执行到这一步时key已经到第二行了即 1001　　01　　1
            context.write(key, NullWritable.get()); setpanme，再写出去
        }

    }
}
遍历value时，key会变；  key像一个杯子，先取出水pname放一个string里，再next下，里边就是装的其他内容；再把pname装进去；最后把这个杯子整体倒出来到一个框架。

View Code