@Override

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

        String[] fields = value.toString().split(" ");

        if (fields.length > 11){ //过滤掉log长度< 11的；

            context.write(value, NullWritable.get());

            context.getCounter("ETL", "true").increment(1);

        }else {

            /*ETL

            false=849

            true=13770

            */

            context.getCounter("ETL", "false").increment(1);

        }

在LogDriver.java中添加：

        job.setMapperClass(LogMapper.class);

        job.setNumReduceTasks(0);

为了支持多种压缩/解压缩算法，Hadoop引入了编码/解码器，如下：

压缩格式    对应的编码/解码器

DEFLATE    org.apache.hadoop.io.compress.DefaultCodec

gzip    org.apache.hadoop.io.compress.GzipCodec

bzip2    org.apache.hadoop.io.compress.BZip2Codec

LZO    com.hadoop.compression.lzo.LzopCodec

Snappy    org.apache.hadoop.io.compress.SnappyCodec

压缩性能的比较

压缩算法    原始文件大小    压缩文件大小    压缩速度    解压速度

gzip    　　.3GB    .8GB    .5MB/s    58MB/s

bzip2  　　  .3GB    .1GB    .4MB/s    .5MB/s

LZO    　　.3GB    .9GB    .3MB/s    .6MB/s

参数mapreduce.map.output.compress（在mapred-site.xml中配置）    默认false    在mapper输出阶段    这个参数设为true启用压缩

mapreduce.map.output.compress.codec（在mapred-site.xml中配置）  默认org.apache.hadoop.io.compress.DefaultCodec    mapper输出    企业多使用LZO或Snappy编解码器在此阶段压缩数据

编码格式可写gzip格式，默认用的snappy但现hadoop不支持

mapreduce.output.fileoutputformat.compress（在mapred-site.xml中配置）    false    reducer输出    这个参数设为true启用压缩

mapreduce.output.fileoutputformat.compress.codec（在mapred-site.xml中配置） org.apache.hadoop.io.compress. DefaultCodec reducer输出 使用标准工具或者编解码器，如gzip和bzip2

mapreduce.output.fileoutputformat.compress.type（在mapred-site.xml中配置）    RECORD    reducer输出    SequenceFile输出使用的压缩类型：NONE和BLOCK

public class TestCompress {

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

        compress("F://web.log", BZip2Codec.class);  //压缩

        //decompress("F://web.log.bz2");  //解压缩

    }                           //传入一个压缩类，通过这个类反射进来一个实例codec，通过实例把流包装起来

    private static void compress(String path, Class<? extends CompressionCodec> codecClass) throws IOException {

        CompressionCodec codec = ReflectionUtils.newInstance(codecClass, new Configuration());

        String extension = codec.getDefaultExtension(); //压缩格式为默认的扩展名，默认的为.bz2;与上边的BZip2Codec.class对应的

                /*public String getDefaultExtension() {

                    return ".bz2";

                }*/

        //开流

        FileInputStream fis = new FileInputStream(path);

        FileOutputStream fos = new FileOutputStream(path + extension);//加个扩展名

        CompressionOutputStream cos = codec.createOutputStream(fos); //包装流，创建输出流的压缩流

        //流要对口

        IOUtils.copyBytes(fis, cos, 1024);

        IOUtils.closeStream(fis);

        IOUtils.closeStream(cos);

    }

    private static void decompress(String path) throws IOException {

        CompressionCodecFactory codecFactory = new CompressionCodecFactory(new Configuration());

        //工厂模式，通过文件名path，getCodec返回一个具体的类，这叫工厂模式，在编码写代码的时候很好用；没工厂模式就要不断的判断if或switch

        CompressionCodec codec = codecFactory.getCodec(new Path(path));

        FileInputStream fis = new FileInputStream(path);

        FileOutputStream fos = new FileOutputStream("F://1.log");

        CompressionInputStream cis = codec.createInputStream(fis); //输出的就不是压缩的了

        //谁被压缩就包谁，这里得是cis

        IOUtils.copyBytes(cis, fos, 1024);

        IOUtils.closeStream(cis);

        IOUtils.closeStream(fis);

    }

}

在Driver.java

 　　　　Configuration configuration = new Configuration();

        // 开启map端输出压缩

        configuration.setBoolean("mapreduce.map.output.compress", true);

        // 设置map端输出压缩方式

        configuration.setClass("mapreduce.map.output.compress.codec", BZip2Codec.class, CompressionCodec.class);

　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　//在工作中不能采用Bzip压缩，文件大点 2M/S的速度，会很慢！！

        //1.获取一个任务实例; 获取配置信息和封装任务

        Job job = Job.getInstance(configuration); //new Configuration()
--->结果看不出什么区别的
 INFO [org.apache.hadoop.io.compress.CodecPool] - Got brand-new compressor [.bz2] ##在shuffle阶段采用压缩

　　　　// 设置reduce端输出压缩开启

        FileOutputFormat.setCompressOutput(job, true);

        // 设置压缩的方式

        FileOutputFormat.setOutputCompressorClass(job, BZip2Codec.class);

        //6.提交任务

        boolean b = job.waitForCompletion(true);

--->最后的结果是part-r-00000.bz2格式的

<property>

      <name>mapreduce.map.speculative</name>

      <value>true</value>

      <description>If true, then multiple instances of some map tasks may be executed in parallel.</description>

</property>

<property>

      <name>mapreduce.reduce.speculative</name>

      <value>true</value>

      <description>If true, then multiple instances of some reduce tasks may be executed in parallel.</description>

</property>