FileInputFormat

FileInputFormat是基本的数据读取类型，包括TextInputFormat、KeyValueInputFormat、NLineInputFormat、CombineTextInputFormat以及自定义的InputFormat。

• TextInputFormat：默认的类型，key是偏移量Long类型，value是一行的数据；
• KeyValueInputFormat：默认以tab分割，一行数据中tab前是key，tab后面是value；
• NLineInputFormat：按行数定义切片大小；
• CombineTextInputFormat：小文件数量较多时会使用，将多个小文件从逻辑上规划到一个切片中。

shuffle机制

Map方法之后，Reduce方法之前的数据处理过程称为Shuffle或洗牌。

partition分区

默认分区是根据key的hashCode对ReduceTasks个数取模得到，用户没法控制哪个key存储到哪个分区。

通过自定义Partitioner，可以实现自定义分区。

例如，按手机号码前三位进行分区。

添加类继承Partitioner，

package partition;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;
/**
 * @author Administrator
 */
public class ProvincePartitioner extends Partitioner<Text, FlowBean> {
    @Override
    public int getPartition(Text text, FlowBean flowBean, int i) {
        String prePhoneNum = text.toString().substring(0, 3);
        int partition=4;
        if ("130".equals(prePhoneNum)) {
            partition = 0;
        } else if ("131".equals(prePhoneNum)) {
            partition = 1;
        } else if ("132".equals(prePhoneNum)) {
            partition = 2;
        } else if ("133".equals(prePhoneNum)) {
            partition = 3;
        }
        return partition;
    }
}

然后在驱动类中添加

job.setPartitionerClass(ProvincePartitioner.class);
job.setNumReduceTasks(5);//默认值是1

这里一共分了5个区，分区号必须从零开始编号、逐一累加。

WritableComparable排序

MapTask和ReduceTask都会对数据按照key进行排序，属于Hadoop的默认行为。

默认排序是按照字典排序，且实现该排序的方法是快速排序。

排序的分类：部分排序、全排序、辅助排序（分组排序）、二次排序。

排序的实现：bean对象作为key传输，实现WritableComparable接口重写compareTo方法。

全排序

全排序要在实例中实现WritableComparable接口重写compareTo方法。

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

分区排序

即分区和排序的结合。

Combiner合并

Combiner是Reducer的子类，在每一个MapTask所在节点运行。意义在于对每一个MapTask的输出进行局部汇总，以减少网络传输量。不适应于求均值的场景，会影响加权。

辅助排序

再自定义类继承WritableComparator类，重写compare方法，并在启动类关联。

以下为例

package group;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
/**
 * @author Administrator
 */
public class OrderGroupComparator extends WritableComparator {
    protected OrderGroupComparator(){
        super(Order.class,true);
    }
    @Override
    public int compare(WritableComparable a, WritableComparable b) {
        Order aBean= (Order) a;
        Order bBean= (Order) b;
        int result;
        if (aBean.getId()>bBean.getId()){
            result=1;
        }else if (aBean.getId()<bBean.getId()){
            result=-1;
        }else {
            result=0;
        }
        return result;
    }
}

OutputFormat

OutputFormat是MapReduce输出的基类，默认值是TextOutputFormat，即把结果记录为文本。

• TextOutputFormat：文本输出；
• SequenceFileOutputFormat：输出结果作为后续MapReduce任务的输入；
• 自定义OutputFormat：包括输出到MySQL、Redis、HDFS等。

压缩与解压

这里以BZip2和Gzip压缩格式为例，直接上代码。

package compress;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
import org.apache.hadoop.io.compress.CompressionInputStream;
import org.apache.hadoop.io.compress.CompressionOutputStream;
import org.apache.hadoop.util.ReflectionUtils;
import java.io.*;
/**
 * @author Administrator
 */
public class TestCompress {
    public static void main(String[] args) throws IOException, ClassNotFoundException {
        //压缩
        //compress("C:/Users/Administrator/Desktop/input/hello.txt","org.apache.hadoop.io.compress.BZip2Codec");
        //compress("C:/Users/Administrator/Desktop/input/hello.txt","org.apache.hadoop.io.compress.GzipCodec");
        //解压
        decompress("C:/Users/Administrator/Desktop/input/hello.txt.gz");
    }
    private static void compress(String fileName, String method) throws IOException, ClassNotFoundException {
        //获取输入输出流
        FileInputStream fileInputStream = new FileInputStream(new File(fileName));
        Class aClass = Class.forName(method);
        CompressionCodec codec = (CompressionCodec) ReflectionUtils.newInstance(aClass, new Configuration());
        FileOutputStream fileOutputStream = new FileOutputStream(new File(fileName + codec.getDefaultExtension()));
        CompressionOutputStream codecOutputStream = codec.createOutputStream(fileOutputStream);
        //流的拷贝
        IOUtils.copyBytes(fileInputStream,codecOutputStream,1024*1024,false);
        //关闭流
        IOUtils.closeStream(codecOutputStream);
        IOUtils.closeStream(fileOutputStream);
        IOUtils.closeStream(fileInputStream);
    }
    private static void decompress(String fileName) throws IOException {
        //压缩方式检查
        CompressionCodecFactory factory = new CompressionCodecFactory(new Configuration());
        CompressionCodec codec = factory.getCodec(new Path(fileName));
        if (codec==null){
            System.out.println("can't process");
            return;
        }
        //获取输入输出流
        FileInputStream fileInputStream = new FileInputStream(new File(fileName));
        CompressionInputStream codecInputStream = codec.createInputStream(fileInputStream);
        FileOutputStream fileOutputStream = new FileOutputStream(new File(fileName + ".decode"));
        //流的对拷
        IOUtils.copyBytes(codecInputStream,fileOutputStream,1024*1024,false);
        //关闭流
        IOUtils.closeStream(fileOutputStream);
        IOUtils.closeStream(codecInputStream);
        IOUtils.closeStream(fileInputStream);
    }
}