【Hadoop】Hadoop数据压缩

【Hadoop】Hadoop数据压缩

文章目录

​​一、压缩概述​​​​二、压缩策略和原则​​​​三、MR支持的压缩编码​​​​四、压缩方式选择​​

​​☼ Gzip压缩​​​​☼ Bzip2压缩​​​​☼Lzo压缩​​​​☼ Snappy压缩​​

​​五、压缩位置选择​​​​六、压缩参数配置​​

​​☼ 输入阶段​​​​☼ Mapper输出阶段​​​​☼ Reducer输出阶段​​

​​七、压缩实操案例​​

​​☼ 数据流的压缩和解压缩​​

​​压缩案例 ---- org.apache.hadoop.io.compress.BZip2Codec​​​​压缩案例 ---- org.apache.hadoop.io.compress.GzipCodec​​​​压缩案例 ---- org.apache.hadoop.io.compress.DefaultCodec​​​​解压缩案例 ---- org.apache.hadoop.io.compress.BZip2Codec/GzipCodec/DefaultCodec​​

​​☼ Map输出端采用压缩 ---- wordCount​​

​​Mapper阶段​​​​Reducer阶段​​​​Driver阶段​​

​​☼ Reduce输出端采用压缩 ---- wordCount​​

一、压缩概述

压缩技术能够有效减少底层存储系统(HDFS)读写字节数,压缩提高了网络带宽和磁盘空间的效率。在运行MR程序时,​​I/O操作​​、​​网络数据传输​​、 ​​Shuffle​​和 ​​Merge​​要花大量的时间,尤其是数据规模很大和工作负载密集的情况下,因此,使用数据压缩显得非常重要。

鉴于磁盘I/O和网络带宽是 Hadoop的贵资源,数据压缩对于节省资源、最小化磁盘I/O和网络传输非常有帮助。​​可以在任意 MapReduce 阶段启用压缩​​。不过,尽管压缩与解压操作的CPU开销不高,其性能的提升和资源的节省并非没有代价。

​​返回顶部​​

二、压缩策略和原则

压缩是提高 Hadoop运行效率的一种优化策略。通过对 Mapper、 Reducer运行过程的数据进行压缩,以减少磁盘IO,提高MR程序运行速度。注意:采用压缩技术减少了磁盘IO,但同时增加了CPU运算负担。所以,压缩特性运用得当能提高性能,但运用不当也可能降低性能。压缩基本原则:

​​运算密集型的job,少用压缩​​​​IO密集型的job,多用压缩​​

​​返回顶部​​

三、MR支持的压缩编码

​​返回顶部​​

四、压缩方式选择

☼ Gzip压缩

优点:

​​压缩率比较高​​,而且压缩/解压速度也比较快;​​Hadoop本身支持​​,在应用中处理Gzip格式的文件就和直接处理文本一样;​​大部分 Linux系统都自带Gzip命令,使用方便​​。

缺点：

​​不支持 Split​​。

应用场景:

​​当每个文件压缩之后在130M以内的(1个块大小内),都可以考虑用Gzip压缩格式​​。例如说一天或者一个小时的日志压缩成一个Gzip文件

​​返回顶部​​

☼ Bzip2压缩

优点:

​​支持 Split​​;​​具有很高的压缩率​​,比Gzi压缩率都高;​​Hadoop本身自带​​,使用方便。

缺点:

​​压缩/解压速度慢​​。

应用场景:

​​适合对速度要求不高,但需要较高的压缩率​​的时候;或者​​输出之后的数据比较大​​​,处理之后的数据需要压缩存档减少磁盘空间并且​​以后数据用得比较少的情况​​;或者​​对单个很大的文本文件想压缩减少存储空间,同时又需要支持 Split,而且兼容之前的应用程序​​的情况。

​​返回顶部​​

☼Lzo压缩

优点:

压缩/解压速度也比较快,​​合理的压缩率​​;​​支持 Split​​,是 Hadoop中最流行的压缩格式;​​可以在 Linux系统下安装lzop命令​​,使用方便

缺点:

​​压缩率比Gzip要低一些​​;​​Hadoop本身不支持​​,需要安装;在应用中对Lzo格式的文件需要做一些特殊处理(​​为了支持 Split需要建索引,还需要指定InputFormat为Lzo格式​​)

应用场景:

一个很大的文本文件,压缩之后还大于200M以上的可以考虑,而且单个文件越大,Lzo优点越越明显。

​​返回顶部​​

☼ Snappy压缩

优点:

​​高速压缩速度和合理的压缩率​​

缺点:

​​不支持Split​​;​​压缩率比Gzip要低​​;​​Hadoop本身不支持​​,需要安装

应用场景:

当 MapReduce作业的​​Map输出的数据比较大的时候​​,作为Map到 Reduce的中间数据的压缩格式;或者​​作为一个 MapReduce作业的输出和另外一个MapReduce作业的输入​​。

​​返回顶部​​

五、压缩位置选择

之前有说到，压缩可以在MR的任意阶段启用，那么如何选取压缩位置也很重要~

​​返回顶部​​

六、压缩参数配置

☼ 输入阶段

☼ Mapper输出阶段

☼ Reducer输出阶段

​​返回顶部​​

七、压缩实操案例

☼ 数据流的压缩和解压缩

CompressionCodec有两个方法可以用于轻松地压缩或解压缩数据。

要想对正在被写入一个输出流的数据进行压缩,我们可以使用​​createOutputStream(OutputStreamout)​​方法创建一个 ​​CompressionOutputStream​​,将其以压缩格式写入底层的流。相反,要想对从输入流读取而来的数据进行解压缩,则调用​​createInputStream(InputStreamin)​​函数,从而获得一个 ​​CompressionInputStream​​,从而从底层的流读取未压缩的数据。

压缩案例 ---- org.apache.hadoop.io.compress.BZip2Codec

package 第三章_MR框架原理.压缩解压缩;import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.io.IOUtils;import org.apache.hadoop.io.compress.CompressionCodec;import org.apache.hadoop.io.compress.CompressionOutputStream;import org.apache.hadoop.util.ReflectionUtils;import java.io.*;public class TestCompress { public static void main(String[] args) throws IOException, ClassNotFoundException { // 压缩 --- 压缩什么？压缩成什么格式？ compress("G:\\学习\\MapReduce\\data\\web.log","org.apache.hadoop.io.compress.BZip2Codec"); } /** * 压缩 * @param fileName * @param method * @throws IOException * @throws ClassNotFoundException */ private static void compress(String fileName,String method) throws IOException, ClassNotFoundException { // 1.获取输出流 FileInputStream fis = new FileInputStream(new File(fileName)); // 2.获取输入流 // 2.1 获取压缩形式后缀名 Class classCodec = Class.forName(method); CompressionCodec codec = (CompressionCodec) ReflectionUtils.newInstance(classCodec, new Configuration()); // 2.2 获取输入流 FileOutputStream fos = new FileOutputStream(new File(fileName + codec.getDefaultExtension())); // 2.3 包装成压缩输出流 CompressionOutputStream cos = codec.createOutputStream(fos); // 3.流的对拷 // false --- 是否关闭输入、输出流 IOUtils.copyBytes(fis,cos,1024*1024,false); // 4.关闭流资源 IOUtils.closeStream(cos); IOUtils.closeStream(fos); IOUtils.closeStream(fis); }}

​​返回顶部​​

压缩案例 ---- org.apache.hadoop.io.compress.GzipCodec

只需要修改method参数即可！！！

public static void main(String[] args) throws IOException, ClassNotFoundException { // 压缩 --- 压缩什么？压缩成什么格式？ //compress("G:\\学习\\MapReduce\\data\\web.log","org.apache.hadoop.io.compress.BZip2Codec"); compress("G:\\学习\\MapReduce\\data\\web.log","org.apache.hadoop.io.compress.GzipCodec");}

​​返回顶部​​

压缩案例 ---- org.apache.hadoop.io.compress.DefaultCodec

public static void main(String[] args) throws IOException, ClassNotFoundException { // 压缩 --- 压缩什么？压缩成什么格式？ //compress("G:\\学习\\MapReduce\\data\\web.log","org.apache.hadoop.io.compress.BZip2Codec"); //compress("G:\\学习\\MapReduce\\data\\web.log","org.apache.hadoop.io.compress.GzipCodec"); compress("G:\\学习\\MapReduce\\data\\web.log","org.apache.hadoop.io.compress.DefaultCodec");}

这里本人的解压工具不支持​​.deflate​​格式解压，大家可以去-有关加压软件试试~

​​返回顶部​​

解压缩案例 ---- org.apache.hadoop.io.compress.BZip2Codec/GzipCodec/DefaultCodec

package 第三章_MR框架原理.压缩解压缩;import com.jcraft.jsch.IO;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 java.io.*;public class TestDeCompress { public static void main(String[] args) throws IOException { //deCompress("G:\\学习\\MapReduce\\data\\web.log.bz2"); //deCompress("G:\\学习\\MapReduce\\data\\web.log.defalte"); deCompress("G:\\学习\\MapReduce\\data\\web.log.gz"); } /** * 解压缩 * @param fileName */ private static void deCompress(String fileName) throws IOException { // 1.压缩方式检查 // 1.1 通过factory获取文件的压缩方式 CompressionCodecFactory factory = new CompressionCodecFactory(new Configuration()); CompressionCodec codec = factory.getCodec(new Path(fileName)); // 1.2 判断文件路径是否正确 if (codec == null){ System.out.println("cannot find codec for file " + fileName); return; } // 1.3 获取输入流 FileInputStream fis = new FileInputStream(new File(fileName)); // 1.3.1 将输入流包装成压缩输入流 CompressionInputStream cis = codec.createInputStream(fis); // 获取输出流 FileOutputStream fos = new FileOutputStream(new File(fileName+".decode")); // 对拷流 IOUtils.copyBytes(cis,fos,1024*1024,false); // 关闭流资源 IOUtils.closeStream(fos); IOUtils.closeStream(cis); IOUtils.closeStream(fis); }}

可以看到最终生成了对应的解压文件，这里本人跳了一步，就是最后将解压的文件重新命名了。

注意：在解压 ​​.bz2​​ 文件的时候报了个错：​​Exception in thread "main" java.io.IOException: Stream is not BZip2 formatted: expected 'h' as first byte but got ''​​，网上没找到详细说明，估计是解压插件不到位（有懂的大佬可以在评论区留言哦~）

​​返回顶部​​

☼ Map输出端采用压缩 ---- wordCount

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

Mapper阶段

package 第三章_MR框架原理.压缩解压缩;import org.apache.hadoop.io.IntWritable;import org.apache.hadoop.io.LongWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapreduce.Mapper;import java.io.IOException;/** * Mapper 阶段 * KEYIN 输入数据的key类型 * VALUEIN 输入数据的value类型 * KEYOUT 输出数据的key类型 * VALUEOUT 输出数据的value类型 */public class wordCountMapper extends Mapper { // 创建对象 Text k = new Text(); IntWritable v = new IntWritable(); @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { // 1.获取一行数据 // atguigu atguigu String line = value.toString(); // 2.切分 String[] words = line.split(" "); // 3.循环写出 for (String word:words){ // 设置键 atguigu k.set(word); // 设置词频为 1 , 也可以在上面创建对象时默认为1 v.set(1); // 生成键值对 (atguigu,1) context.write(k,v); } }}

​​返回顶部​​

Reducer阶段

package 第三章_MR框架原理.压缩解压缩;import org.apache.hadoop.io.IntWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapreduce.Reducer;import java.io.IOException;/** * Reducer 阶段 * KEYIN ,VALUEIN Reducer阶段输入（Mapper阶段输出）数据的类型 * KEYOUT 最终输出数据的key类型 * VALUEOUT 最终输出数据的value类型 */public class wordCountReducer extends Reducer { IntWritable v = new IntWritable(); @Override // Iterable values 对key的value值进行迭代实现词频统计 protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException { // atguigu，1 // atguigu，1 // 1.累加求和 int sum = 0; for (IntWritable value:values){ // value是IntWritable类型数据，通过get转为int型，才好计算 sum += value.get(); } // 2.写出结果 v.set(sum); context.write(key,v); }}

​​返回顶部​​

Driver阶段

package 第三章_MR框架原理.压缩解压缩;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.mapreduce.Job;import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;import java.io.IOException;public class wordCountDriver { public static void main(String[] args) { Configuration conf = new Configuration(); // 开启map端输出压缩 conf.setBoolean("mapreduce.map.output.compress", true); // 设置map端输出压缩方式 conf.setClass("mapreduce.map.output.compress.codec", BZip2Codec.class, CompressionCodec.class); Job job = null; try { // 1.获取job对象 job = Job.getInstance(conf); // 2.设置jar存储位置 job.setJarByClass(wordCountDriver.class); // 3.关联map、reduce类 job.setMapperClass(wordCountMapper.class); job.setReducerClass(wordCountReducer.class); // 4.设置Mapper阶段输出数据的key、value类型 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(IntWritable.class); // 5.设置Reducer阶段输出数据的key、value类型 job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); // 6.设置输入、出路径 FileInputFormat.setInputPaths(job,new Path("G:\\Projects\\IdeaProject-C\\MapReduce\\src\\main\\java\\第三章_MR框架原理\\压缩解压缩\\dataset\\")); FileOutputFormat.setOutputPath(job,new Path("G:\\Projects\\IdeaProject-C\\MapReduce\\src\\main\\java\\第三章_MR框架原理\\压缩解压缩\\output\\")); // 7.提交job job.waitForCompletion(true); } catch (IOException e) { e.printStackTrace(); } catch (InterruptedException e) { e.printStackTrace(); } catch (ClassNotFoundException e) { e.printStackTrace(); } }}

变动：只需要在驱动类中开启map阶段的压缩，并设置压缩类型就可以了

// 开启map端输出压缩conf.setBoolean("mapreduce.map.output.compress", true);// 设置map端输出压缩方式conf.setClass("mapreduce.map.output.compress.codec", BZip2Codec.class, CompressionCodec.class);

​​返回顶部​​

☼ Reduce输出端采用压缩 ---- wordCount

package 第三章_MR框架原理.压缩解压缩;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.mapreduce.Job;import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;import java.io.IOException;public class wordCountDriver { public static void main(String[] args) { Configuration conf = new Configuration(); // 开启map端输出压缩 conf.setBoolean("mapreduce.map.output.compress", true); // 设置map端输出压缩方式 conf.setClass("mapreduce.map.output.compress.codec", BZip2Codec.class, CompressionCodec.class); Job job = null; try { // 1.获取job对象 job = Job.getInstance(conf); // 2.设置jar存储位置 job.setJarByClass(wordCountDriver.class); // 3.关联map、reduce类 job.setMapperClass(wordCountMapper.class); job.setReducerClass(wordCountReducer.class); // 4.设置Mapper阶段输出数据的key、value类型 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(IntWritable.class); // 5.设置Reducer阶段输出数据的key、value类型 job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); // 设置reduce端输出压缩开启 FileOutputFormat.setCompressOutput(job, true); // 设置压缩的方式 FileOutputFormat.setOutputCompressorClass(job, BZip2Codec.class); // 6.设置输入、出路径 FileInputFormat.setInputPaths(job,new Path("G:\\Projects\\IdeaProject-C\\MapReduce\\src\\main\\java\\第三章_MR框架原理\\压缩解压缩\\dataset\\")); FileOutputFormat.setOutputPath(job,new Path("G:\\Projects\\IdeaProject-C\\MapReduce\\src\\main\\java\\第三章_MR框架原理\\压缩解压缩\\output\\")); // 7.提交job job.waitForCompletion(true); } catch (IOException e) { e.printStackTrace(); } catch (InterruptedException e) { e.printStackTrace(); } catch (ClassNotFoundException e) { e.printStackTrace(); } }}

改动

// 设置reduce端输出压缩开启FileOutputFormat.setCompressOutput(job, true);// 设置压缩的方式FileOutputFormat.setOutputCompressorClass(job, BZip2Codec.class);

​​返回顶部​​

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。