什么是MapReduce

MapReduce是一个分布式运算程序的编程框架，MapReduce核心功能是将用户编写的业务逻辑代码和自带默认组件整合成一个完整的分布式运算程序，并发运行在一个Hadoop集群上。

MapReduce优缺点

优点

  1. 易于编程。
  2. 良好的扩展性(基于Hadoop)
  3. 容错性高。其中一台机器挂了，它会把上面的计算任务转移到另一台节点上运行。

  4. 适合PB级以上海量数据离线处理。

    缺点

  5. 不擅长实时计算。

  6. 不擅长流式计算。

  7. 不擅长DAG计算。

    MapReduce进程

    一个完整的MapReduce程序在分布式运行时有三类实例进程

  8. MRAppMaster: 负责整个程序的过程调度及状态协调。

  9. MapTask: 负责Map阶段的整个数据处理流程。

  10. ReduceTask: 负责Reduce阶段的整个数据处理流程。

    常用数据序列化类型

    image.png

    MapReduce编程规范

    用户编写的程序分成三个部分:Mapper，Reducer和Driver。

    Mapper阶段

  11. 用户自定义的Mapper要继承自己的父类。

  12. Mapper的输入数据是KV对的形式(KV的类型可以自定义)。
  13. Mapper中的业务逻辑写在map()方法中。
  14. Mapper的输出数据是KV对的形式(KV的类型可自定义)。

  15. map()方法(MapTask进程)对每一个调用一次。

    Reduce阶段

  16. 用户自定的Reducer要继承自己的父类

  17. Reducer的输入数据类型对应Mapper的输出数据类型，也是KV
  18. Reducer的业务逻辑写在reduce()方法中。

  19. ReduceTask进程对每一组相同K的组调用一次reduce()方法。

    Driver阶段

    相当于YARN集群的客户端，用于提交我们整个程序到YARN集群，提交的是封装了MapReduce程序相关运行参数的job对象。

    WorldCount案例实操

    创建maven工程并添加相应依赖

    1. <dependencies>
    2.  <dependency>
    3.      <groupId>junit</groupId>
    4.      <artifactId>junit</artifactId>
    5.      <version>4.12</version>
    6.  </dependency>
    7.  <dependency>
    8.      <groupId>org.apache.logging.log4j</groupId>
    9.      <artifactId>log4j-slf4j-impl</artifactId>
    10.      <version>2.12.0</version>
    11.  </dependency>
    12.  <dependency>
    13.      <groupId>org.apache.hadoop</groupId>
    14.      <artifactId>hadoop-client-api</artifactId>
    15.      <version>3.1.3</version>
    16.  </dependency>
    17.  <dependency>
    18.      <groupId>org.apache.hadoop</groupId>
    19.      <artifactId>hadoop-client-runtime</artifactId>
    20.      <version>3.1.3</version>
    21.  </dependency>
    22. </dependencies>

    在项目的src/main/resources目录下，新建一个文件，命名为“log4j.xml”，在文件中填入。 ```xml <?xml version=”1.0” encoding=”UTF-8”?> 

<a name="HDVAY"></a>
#### 编写Mapper类
```java
import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class WordcountMapper extends Mapper<LongWritable, Text, Text, IntWritable>{

    Text k = new Text();
    IntWritable v = new IntWritable(1);

    @Override
    protected void map(LongWritable key, Text value, Context context)    throws IOException, InterruptedException {

        // 1 获取一行
        String line = value.toString();

        // 2 切割
        String[] words = line.split(" ");

        // 3 输出
        for (String word : words) {

            k.set(word);
            context.write(k, v);
        }
    }
}

编写Reducer类

import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class WordcountReducer extends Reducer<Text, IntWritable, Text, IntWritable>{

int sum;
IntWritable v = new IntWritable();

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

        // 1 累加求和
        sum = 0;
        for (IntWritable count : values) {
            sum += count.get();
        }

        // 2 输出
       v.set(sum);
        context.write(key,v);
    }
}

编写Driver驱动类

import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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;

public class WordcountDriver {

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

        // 1 获取配置信息以及封装任务
        Configuration configuration = new Configuration();
        Job job = Job.getInstance(configuration);

        // 2 设置jar加载路径
        job.setJarByClass(WordcountDriver.class);

        // 3 设置map和reduce类
        job.setMapperClass(WordcountMapper.class);
        job.setReducerClass(WordcountReducer.class);

        // 4 设置map输出
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        // 5 设置最终输出kv类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        // 6 设置输入和输出路径
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        // 7 提交
        boolean result = job.waitForCompletion(true);

        System.exit(result ? 0 : 1);
    }
}

集群上测试

用maven打jar包，需要添加的打包插件依赖

<build>
        <plugins>
            <plugin>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>2.3.2</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>
            <plugin>
                <artifactId>maven-assembly-plugin </artifactId>
                <configuration>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                    <archive>
                        <manifest>
                           <!-- 替换为自己工程主类 -->
                            <mainClass>com.mr.WordcountDriver</mainClass>
                        </manifest>
                    </archive>
                </configuration>
                <executions>
                    <execution>
                        <id>make-assembly</id>
                        <phase>package</phase>
                        <goals>
                            <goal>single</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

执行WorldCount程序

hadoop jar  wordcount.jar com.mr.wordcount.WordcountDriver /user/input /user/output

在Windows上向集群提交任务

添加必要配置

public class WordcountDriver {

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

        // 1 获取配置信息以及封装任务
        Configuration configuration = new Configuration();

       configuration.set("fs.defaultFS", "hdfs://hadoop102:8020");
       configuration.set("mapreduce.framework.name","yarn");
       configuration.set("mapreduce.app-submission.cross-platform","true");
    configuration.set("yarn.resourcemanager.hostname","hadoop103");

        Job job = Job.getInstance(configuration);

        // 2 设置jar加载路径
        job.setJarByClass(WordcountDriver.class);
    // job.setJar("D:\\worldcount.jar");

        // 3 设置map和reduce类
        job.setMapperClass(WordcountMapper.class);
        job.setReducerClass(WordcountReducer.class);

        // 4 设置map输出
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        // 5 设置最终输出kv类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        // 6 设置输入和输出路径
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        // 7 提交
        boolean result = job.waitForCompletion(true);

        System.exit(result ? 0 : 1);
    }
}

Hadoop序列化

什么是序列化

序列化就是把内存中的对象，转换成字节序列(或其他传输协议),利于存储到磁盘和网络传输。
反序列化就是将收到字节序列(或其他数据传输协议)或是磁盘的持久化数据，转成内存中的对象。

为什么要序列化

一般对象只生存在内存里，关机断电就没有了。而且对象只能由本地的进程使用，不能被发送到网络上的另外一台计算机。序列化可以做到。

为什么不用Java的序列化

Java序列化是一个重量级序列化框架(Serializable), 一个对象序列化后会附带很多额外的信息(各种校验信息，Header，继承体系等),不便于在网络中高效传输。所以Hadoop自己开发了一套序列化机制(Writable)。

Hadoop序列化特点

  1. 紧凑:高效使用存储空间。
  2. 快速: 读写数据的额外开销小。
  3. 可扩展: 随着通信协议的升级而升级。

  4. 互操作: 支持多语言的交互。

    自定义bean对象实现序列化接口(writable)

  5. 必须实现Writable接口

  6. 反序列化时，需要反射调用空参构造函数，所以必须有空参构造

    public FlowBean() {
 super();
}

  7. 重写序列化方法

    @Override
public void write(DataOutput out) throws IOException {
 out.writeLong(upFlow);
 out.writeLong(downFlow);
 out.writeLong(sumFlow);
}

  8. 重写反序列化方法

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

  9. 注意反序列化的顺序和序列化的顺序完全一致

  10. 要想把结果显示在文件中，需要重写toString()，可用”\t”分开，方便后续用。
  11. 如果需要将自定义的bean放在key中传输，则还需要实现Comparable接口，因为MapReduce框中的Shuffle过程要求对key必须能排序
    @Override
public int compareTo(FlowBean o) {
 // 倒序排列，从大到小
 return this.sumFlow > o.getSumFlow() ? -1 : 1;
}

    MapReduce框架原理

image.png

切片与MapTask并行度决定机制

MapTask的并行度决定Map阶段的任务处理并发度，进而影响到整个Job的处理速度。
数据块：Block是HDFS物理上把数据分成一块一块。
数据切片：数据切片只是在逻辑上对输入进行分片，并不会在磁盘上将其切分成片进行存储。
1.一个Job的Map阶段并行度由客户端在提交Job时的切片数决定
2.每个Split切片分配一个MapTask并行实例处理。
3.默认情况下，切片大小=BlockSize
4.切片时不考虑数据集整体，而是逐个针对每个文件单独切片

CombineTextInputFormat切片机制

框架默认的TextInputFormat切片机制是对任务按文件规划切片，不管文件多小，都会是一个单独的切片，都会交给一个MapTask，这样如果有大量小文件，就会产生大量的MapTask，处理效率极其低下。

  1. 应用场景

CombineTextInputFormat用于小文件过多的场景，它可以将多个小文件从逻辑上规划到一个切片中，这样，多个小文件就可以交给一个MapTask处理。

  1. 虚拟存储切片最大值设置

CombineTextInputFormat.setMaxInputSplitSize(job, 4194304);// 4m
注意：虚拟存储切片最大值设置最好根据实际的小文件大小情况来设置具体的值。

  1. 切片机制

生成切片过程包括：虚拟存储过程和切片过程二部分。

自定义InputFormat案例实操

  1. 自定义一个类继承FileInputFormat
    • 重写isSpliteable()方法，返回false不可切割。
    • 重写createRecordReader(),创建自定义的RecordReader对象，并初始化 ```java import java.io.IOException; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.BytesWritable; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.mapreduce.InputSplit; import org.apache.hadoop.mapreduce.JobContext; import org.apache.hadoop.mapreduce.RecordReader; import org.apache.hadoop.mapreduce.TaskAttemptContext; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

// 定义类继承FileInputFormat public class WholeFileInputformat extends FileInputFormat{

@Override
protected boolean isSplitable(JobContext context, Path filename) {
    return false;
}

@Override
public RecordReader<Text, BytesWritable> createRecordReader(InputSplit split, TaskAttemptContext context)    throws IOException, InterruptedException {

    WholeRecordReader recordReader = new WholeRecordReader();
    recordReader.initialize(split, context);

    return recordReader;
}

}


2. 改写RecordReader
```java
（2）自定义RecordReader类
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

public class WholeRecordReader extends RecordReader<Text, BytesWritable>{

    private Configuration configuration;
    private FileSplit split;

    private boolean isProgress= true;
    private BytesWritable value = new BytesWritable();
    private Text k = new Text();

    @Override
    public void initialize(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {

        this.split = (FileSplit)split;
        configuration = context.getConfiguration();
    }

    @Override
    public boolean nextKeyValue() throws IOException, InterruptedException {

        if (isProgress) {

            // 1 定义缓存区
            byte[] contents = new byte[(int)split.getLength()];

            FileSystem fs = null;
            FSDataInputStream fis = null;

            try {
                // 2 获取文件系统
                Path path = split.getPath();
                fs = path.getFileSystem(configuration);

                // 3 读取数据
                fis = fs.open(path);

                // 4 读取文件内容
                IOUtils.readFully(fis, contents, 0, contents.length);

                // 5 输出文件内容
                value.set(contents, 0, contents.length);

// 6 获取文件路径及名称
String name = split.getPath().toString();

// 7 设置输出的key值
k.set(name);

            } catch (Exception e) {

            }finally {
                IOUtils.closeStream(fis);
            }

            isProgress = false;

            return true;
        }

        return false;
    }

    @Override
    public Text getCurrentKey() throws IOException, InterruptedException {
        return k;
    }

    @Override
    public BytesWritable getCurrentValue() throws IOException, InterruptedException {
        return value;
    }

    @Override
    public float getProgress() throws IOException, InterruptedException {
        return 0;
    }

    @Override
    public void close() throws IOException {
    }
}

  1. 设置Driver ```java // 输入输出路径需要根据自己电脑上实际的输入输出路径设置

     args = new String[] { "e:/input/inputinputformat", "e:/output1" };

// 1 获取job对象
 Configuration conf = new Configuration();
 Job job = Job.getInstance(conf);

// 2 设置jar包存储位置、关联自定义的mapper和reducer
 job.setJarByClass(SequenceFileDriver.class);
 job.setMapperClass(SequenceFileMapper.class);
 job.setReducerClass(SequenceFileReducer.class);

// 7设置输入的inputFormat
 job.setInputFormatClass(WholeFileInputformat.class);

// 8设置输出的outputFormat

    job.setOutputFormatClass(SequenceFileOutputFormat.class);

// 3 设置map输出端的kv类型 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(BytesWritable.class);

   // 4 设置最终输出端的kv类型
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(BytesWritable.class);

   // 5 设置输入输出路径
    FileInputFormat.setInputPaths(job, new Path(args[0]));
    FileOutputFormat.setOutputPath(job, new Path(args[1]));

   // 6 提交job
    boolean result = job.waitForCompletion(true);
    System.exit(result ? 0 : 1);


<a name="QYslx"></a>
### MapReduce工作流程
<a name="lSwak"></a>
#### MapReduce工作流程1
![image.png](https://cdn.nlark.com/yuque/0/2021/png/699900/1632886779278-617a18ac-3799-491a-8237-b920ff456dc7.png#clientId=ub668203e-71fa-4&crop=0&crop=0&crop=1&crop=1&from=paste&height=480&id=u727b8cc8&margin=%5Bobject%20Object%5D&name=image.png&originHeight=480&originWidth=922&originalType=binary&ratio=1&rotation=0&showTitle=false&size=65534&status=done&style=none&taskId=u9fcf2175-f4b8-4616-a98b-793283ec635&title=&width=922)
<a name="MKzQg"></a>
#### MapReduce工作流程2
![image.png](https://cdn.nlark.com/yuque/0/2021/png/699900/1632886867231-16c55115-f632-4248-bbf9-f278e1190f09.png#clientId=ub668203e-71fa-4&crop=0&crop=0&crop=1&crop=1&from=paste&height=453&id=u9a607e4d&margin=%5Bobject%20Object%5D&name=image.png&originHeight=453&originWidth=972&originalType=binary&ratio=1&rotation=0&showTitle=false&size=51662&status=done&style=none&taskId=u6a0b0834-79ae-4b29-a94a-84bf54e85b5&title=&width=972)
<a name="ZUNIE"></a>
#### 流程详解
1.MapTask收集我们的map()方法输出的kv对，放到内存缓冲区中<br />2.从内存缓冲区不断溢出本地磁盘文件，可能会溢出多个文件<br />3.多个溢出文件会被合并成大的溢出文件<br />4.在溢出过程及合并的过程中，都要调用Partitioner进行分区和针对key进行排序<br />5.ReduceTask根据自己的分区号，去各个MapTask机器上取相应的结果分区数据<br />6.ReduceTask会取到同一个分区的来自不同MapTask的结果文件，ReduceTask会将这些文件再进行合并（归并排序）<br />7.合并成大文件后，Shuffle的过程也就结束了，后面进入ReduceTask的逻辑运算过程（从文件中取出一个一个的键值对Group，调用用户自定义的reduce()方法）
<a name="FN0jr"></a>
### Shuffle机制
![image.png](https://cdn.nlark.com/yuque/0/2021/png/699900/1632896206120-22ff0a49-f1c9-4a12-8056-3c7a4299acb1.png#clientId=ub668203e-71fa-4&crop=0&crop=0&crop=1&crop=1&from=paste&height=439&id=u3dd3df85&margin=%5Bobject%20Object%5D&name=image.png&originHeight=439&originWidth=971&originalType=binary&ratio=1&rotation=0&showTitle=false&size=65038&status=done&style=none&taskId=ue6ecdf0f-c34e-4c5c-bc6b-5189b1793b5&title=&width=971)

1.经过mapTask的输出的数据会输入到环形缓存区。<br />2.在环形缓存区中会对KV数据按照规则会对分区排序(快排)。<br />3.当环形缓存区中的数据达到溢写的阈值时，会生成一些小的溢写文件。<br />4.每个MapTask会将上一步生成的小文件进行归并和排序，合成一个大文件。
<a name="tD1RA"></a>
#### 自定义分区器

1. 自定义类继承Partitioner，重写getPartition()方法。
1. 在Job驱动中，设置自定义Partitioner
1. 自定义Partitioner后，要根据自定义Partitioner的逻辑设置相应的数量的ReduceTask
```java
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;

public class ProvincePartitioner extends Partitioner<Text, FlowBean> {

    @Override
    public int getPartition(Text key, FlowBean value, int numPartitions) {

        // 1 获取电话号码的前三位
        String preNum = key.toString().substring(0, 3);

        int partition = 4;

        // 2 判断是哪个省
        if ("136".equals(preNum)) {
            partition = 0;
        }else if ("137".equals(preNum)) {
            partition = 1;
        }else if ("138".equals(preNum)) {
            partition = 2;
        }else if ("139".equals(preNum)) {
            partition = 3;
        }

        return partition;
    }
}

在驱动函数中增加自定义数据分区设置和ReduceTask设置

// 8 指定自定义数据分区
job.setPartitionerClass(ProvincePartitioner.class);

// 9 同时指定相应数量的reduce task
job.setNumReduceTasks(5);

WritableComparable排序

排序是MapReduce框架中最重要的操作之一。MapTask和ReduceTask均会对数据按照key进行排序。该操作属于Hadoop的默认行为。任何应用程序中的数据均会被排序，而不管逻辑上是否需要。
默认排序是按照字典顺序排序，且实现该排序的方法是快速排序。
对于MapTask，它会将处理的结果暂时放到环形缓存区中，当环形缓冲区使用率达到一定阈值后，再对缓冲区中的数据进行一次快速排序，并将这些有序数据溢写到磁盘，而当数据处理完毕后，它会对磁盘上所有文件进行归并排序。
对于ReduceTask，它从每个MapTask上远程拷贝相应的数据文件，如果文件大小超过一定阈值，则溢写磁盘上，否则存储在内存中。如果磁盘上文件数目达到一定阈值，则进行一次归并排序以生成一个更大文件；如果内存文件大小或者数目超过一定阈值，则进行一次合并后将数据溢写到磁盘上。当所有数据拷贝完毕后，ReduceTask统一对内存和磁盘上的所有数据进行一次归并排序。

案例

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import org.apache.hadoop.io.WritableComparable;

public class FlowBean implements WritableComparable<FlowBean> {

    private long upFlow;
    private long downFlow;
    private long sumFlow;

    // 反序列化时，需要反射调用空参构造函数，所以必须有
    public FlowBean() {
        super();
    }

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

    public void set(long upFlow, long downFlow) {
        this.upFlow = upFlow;
        this.downFlow = downFlow;
        this.sumFlow = upFlow + downFlow;
    }

    public long getSumFlow() {
        return sumFlow;
    }

    public void setSumFlow(long sumFlow) {
        this.sumFlow = sumFlow;
    }    

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

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

    /**
     * 反序列化方法 注意反序列化的顺序和序列化的顺序完全一致
     * @param in
     * @throws IOException
     */
    @Override
    public void readFields(DataInput in) throws IOException {
        upFlow = in.readLong();
        downFlow = in.readLong();
        sumFlow = in.readLong();
    }

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

    @Override
    public int compareTo(FlowBean bean) {

        int result;

        // 按照总流量大小，倒序排列
        if (sumFlow > bean.getSumFlow()) {
            result = -1;
        }else if (sumFlow < bean.getSumFlow()) {
            result = 1;
        }else {
            result = 0;
        }

        return result;
    }
}

GroupingComparator分组（辅助排序）

对Reduce阶段的数据根据某一个或几个字段进行分组。
分组排序步骤：

  1. 自定义类继承WritableComparator
  2. 重写compare()方法
    Override
public int compare(WritableComparable a, WritableComparable b) {
     // 比较的业务逻辑
     return result;
}

    案例代码

    ```java import org.apache.hadoop.io.WritableComparable; import org.apache.hadoop.io.WritableComparator;

public class OrderGroupingComparator extends WritableComparator {

protected OrderGroupingComparator() {
    super(OrderBean.class, true);
}

@Override
public int compare(WritableComparable a, WritableComparable b) {

    OrderBean aBean = (OrderBean) a;
    OrderBean bBean = (OrderBean) b;

    int result;
    if (aBean.getOrder_id() > bBean.getOrder_id()) {
        result = 1;
    } else if (aBean.getOrder_id() < bBean.getOrder_id()) {
        result = -1;
    } else {
        result = 0;
    }

    return result;
}

}

设置driver端
```java
    // 8 设置reduce端的分组
    job.setGroupingComparatorClass(OrderGroupingComparator.class);