您的位置: 首页  >  文章  >  Hadoop数据压缩

Hadoop数据压缩

分类: 文章 • 2024-01-04 21:30:34

压缩技术能够要写减少底层存储系统（HDFS）读写字节数。磁盘I/O和网络带宽是Hadoop的宝贵资源，数据压缩对于节省资源，最小化磁盘I/O和网络传输非常有帮助。可以在任意MapReduce阶段启用压缩。

压缩是提高Hadoop运行效率的一种优化策略

通过对Mapper输入，Mapper输出，Reducer输出数据进行压缩，以减少磁盘IO，提高MR程序运行速度。

注意：采用压缩技术减少了磁盘IO，但同时增加了CPU运算负担。所以，压缩特性运用得当能提高性能，但运用不当也可能降低性能。

压缩基本规则：

1）运算密集型的job，少用压缩

2）IO密集型的job，多用压缩

MR支持的压缩编码

表4-7

压缩格式

hadoop自带？

算法

文件扩展名

是否可切分

换成压缩格式后，原来的程序是否需要修改

DEFLATE

是，直接使用

DEFLATE

.deflate

否

和文本处理一样，不需要修改

Gzip

是，直接使用

DEFLATE

.gz

否

和文本处理一样，不需要修改

bzip2

是，直接使用

bzip2

.bz2

是

和文本处理一样，不需要修改

LZO

否，需要安装

LZO

.lzo

是

需要建索引，还需要指定输入格式

Snappy

否，需要安装

Snappy

.snappy

否

和文本处理一样，不需要修改

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

表4-8

压缩格式

对应的编码/解码器

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

压缩性能的比较

表4-9

压缩算法

原始文件大小

压缩文件大小

压缩速度

解压速度

gzip

8.3GB

1.8GB

17.5MB/s

58MB/s

bzip2

8.3GB

1.1GB

2.4MB/s

9.5MB/s

LZO

8.3GB

2.9GB

49.3MB/s

74.6MB/s

http://google.github.io/snappy/

On a single core of a Core i7 processor in 64-bit mode, Snappy compresses at about 250 MB/sec or more and decompresses at about 500 MB/sec or more.

4.3 压缩方式选择

gzip压缩

Hadoop数据压缩

Bzip2压缩

Hadoop数据压缩

Lzo压缩

Hadoop数据压缩

 Snappy压缩

Hadoop数据压缩

压缩位置选择

压缩可以在MapReduce作用的任意阶段启用，如图4-22所示。

Hadoop数据压缩

压缩参数配置

要在Hadoop中启用压缩，可以配置如下参数：

表4-10 配置参数

参数

默认值

阶段

建议

io.compression.codecs  

（在core-site.xml中配置）

org.apache.hadoop.io.compress.DefaultCodec, org.apache.hadoop.io.compress.GzipCodec, org.apache.hadoop.io.compress.BZip2Codec

 

输入压缩

Hadoop使用文件扩展名判断是否支持某种编解码器

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编解码器在此阶段压缩数据

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

数据流的压缩和解压缩

Hadoop数据压缩

测试一下如下压缩方式：

表4-11

DEFLATE

org.apache.hadoop.io.compress.DefaultCodec

gzip

org.apache.hadoop.io.compress.GzipCodec

bzip2

org.apache.hadoop.io.compress.BZip2Codec

package com.atguigu.mapreduce.compress;

import java.io.File;

import java.io.FileInputStream;

import java.io.FileNotFoundException;

import java.io.FileOutputStream;

import java.io.IOException;

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;

 

public class TestCompress {

 

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

       compress("e:/hello.txt","org.apache.hadoop.io.compress.BZip2Codec");

//     decompress("e:/hello.txt.bz2");

    }

 

    // 1、压缩

    private static void compress(String filename, String method) throws Exception {

      

       // （1）获取输入流

       FileInputStream fis = new FileInputStream(new File(filename));

      

       Class codecClass = Class.forName(method);

      

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

      

       // （2）获取输出流

       FileOutputStream fos = new FileOutputStream(new File(filename + codec.getDefaultExtension()));

       CompressionOutputStream cos = codec.createOutputStream(fos);

      

       // （3）流的对拷

       IOUtils.copyBytes(fis, cos, 1024*1024*5, false);

      

       // （4）关闭资源

       cos.close();

       fos.close();

fis.close();

    }

 

    // 2、解压缩

    private static void decompress(String filename) throws FileNotFoundException, IOException {

      

       // （0）校验是否能解压缩

       CompressionCodecFactory factory = new CompressionCodecFactory(new Configuration());

 

       CompressionCodec codec = factory.getCodec(new Path(filename));

      

       if (codec == null) {

           System.out.println("cannot find codec for file " + filename);

           return;

       }

      

       // （1）获取输入流

       CompressionInputStream cis = codec.createInputStream(new FileInputStream(new File(filename)));

      

       // （2）获取输出流

       FileOutputStream fos = new FileOutputStream(new File(filename + ".decoded"));

      

       // （3）流的对拷

       IOUtils.copyBytes(cis, fos, 1024*1024*5, false);

      

       // （4）关闭资源

       cis.close();

       fos.close();

    }

}

4.6.2 Map输出端采用压缩

即使你的MapReduce的输入输出文件都是未压缩的文件，你仍然可以对Map任务的中间结果输出做压缩，因为它要写在硬盘并且通过网络传输到Reduce节点，对其压缩可以提高很多性能，这些工作只要设置两个属性即可，我们来看下代码怎么设置。

1．给大家提供的Hadoop源码支持的压缩格式有：BZip2Codec 、DefaultCodec

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.io.compress.BZip2Codec;

import org.apache.hadoop.io.compress.CompressionCodec;

import org.apache.hadoop.io.compress.GzipCodec;

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 {

 

       Configuration configuration = new Configuration();

 

       // 开启map端输出压缩

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

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

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

 

       Job job = Job.getInstance(configuration);

 

       job.setJarByClass(WordCountDriver.class);

 

       job.setMapperClass(WordCountMapper.class);

       job.setReducerClass(WordCountReducer.class);

 

       job.setMapOutputKeyClass(Text.class);

       job.setMapOutputValueClass(IntWritable.class);

 

       job.setOutputKeyClass(Text.class);

       job.setOutputValueClass(IntWritable.class);

 

       FileInputFormat.setInputPaths(job, new Path(args[0]));

       FileOutputFormat.setOutputPath(job, new Path(args[1]));

 

       boolean result = job.waitForCompletion(true);

 

       System.exit(result ? 1 : 0);

    }

}

2．Mapper保持不变

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

       }

    }

}

3．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>{

 

    IntWritable v = new IntWritable();

 

    @Override

    protected void reduce(Text key, Iterable<IntWritable> values,

           Context context) throws IOException, InterruptedException {

      

       int sum = 0;

 

       // 1 汇总

       for(IntWritable value:values){

           sum += value.get();

       }

      

        v.set(sum);

 

        // 2 输出

       context.write(key, v);

    }

}

4.6.3 Reduce输出端采用压缩

基于WordCount案例处理。

1．修改驱动

 

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.io.compress.BZip2Codec;

import org.apache.hadoop.io.compress.DefaultCodec;

import org.apache.hadoop.io.compress.GzipCodec;

import org.apache.hadoop.io.compress.Lz4Codec;

import org.apache.hadoop.io.compress.SnappyCodec;

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 {

      

       Configuration configuration = new Configuration();

      

       Job job = Job.getInstance(configuration);

      

       job.setJarByClass(WordCountDriver.class);

      

       job.setMapperClass(WordCountMapper.class);

       job.setReducerClass(WordCountReducer.class);

      

       job.setMapOutputKeyClass(Text.class);

       job.setMapOutputValueClass(IntWritable.class);

      

       job.setOutputKeyClass(Text.class);

       job.setOutputValueClass(IntWritable.class);

      

       FileInputFormat.setInputPaths(job, new Path(args[0]));

       FileOutputFormat.setOutputPath(job, new Path(args[1]));

      

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

       FileOutputFormat.setCompressOutput(job, true);

      

       // 设置压缩的方式

        FileOutputFormat.setOutputCompressorClass(job, BZip2Codec.class);

//      FileOutputFormat.setOutputCompressorClass(job, GzipCodec.class);

//      FileOutputFormat.setOutputCompressorClass(job, DefaultCodec.class);

       

       boolean result = job.waitForCompletion(true);

      

       System.exit(result?1:0);

    }

}

 

 

 

 

 

 

 

 

 

相关推荐