hadoop 哪个压缩编码好（BZip2Codec压缩Reduce端压缩控制）

小君 2023-09-15 17:02:27 653

hadoop 哪个压缩编码好（BZip2Codec压缩Reduce端压缩控制）MR在进行压缩数据处理时，不需要手工去处理。但是MR执行过程中，是需要一个动作进行数据压缩、解压的，MR根据后缀名进行处理。在处理压缩、解压的过程中，是有CPU资源的消耗的。如果集群本来CPU使用率很高，就会对其他作业产生影响，不建议使用压缩。大数据处理流程基本上是输入==>处理==>输出，举例来说，在离线处理方面，Spark可以HDFS==>Spark==>HDFS，在实时处理方面，Spark Streaming可以Kafka==>Spark Streaming==>RDBMS。压缩可以使用在输入时，也可以使用在处理时（比如map作为中间状态，它的输出可以压缩以减少Shuffle的量），输出时。今天来聊聊 hadoop 的压缩。压缩：原始数据通过压缩手段产生目标数据，要求输入和输出的内容是一样的（大部分），但体积是不一样的。对于单机用户来说，磁盘空间的

hadoop 哪个压缩编码好（BZip2Codec压缩Reduce端压缩控制）(1)

作者 | Tai_Park

责编 | Carol

来源 | CSDN 博客

封图 | CSDN付费下载于东方 IC

今天来聊聊 hadoop 的压缩。

压缩：原始数据通过压缩手段产生目标数据，要求输入和输出的内容是一样的（大部分），但体积是不一样的。

对于单机用户来说，磁盘空间的限制导致了文件压缩的需求；对于Hadoop用户来说，由于DataNode的限制，也要对HDFS上的数据进行压缩。压缩的目的是减少存储在HDFS上的数据所占用的空间（磁盘的角度），提升网络的传输效率（网络的角度）。对于分布式计算框架来说，Shuffle是一个主要的技术瓶颈。

大数据处理流程基本上是输入==>处理==>输出，举例来说，在离线处理方面，Spark可以HDFS==>Spark==>HDFS，在实时处理方面，Spark Streaming可以Kafka==>Spark Streaming==>RDBMS。压缩可以使用在输入时，也可以使用在处理时（比如map作为中间状态，它的输出可以压缩以减少Shuffle的量），输出时。

MR在进行压缩数据处理时，不需要手工去处理。但是MR执行过程中，是需要一个动作进行数据压缩、解压的，MR根据后缀名进行处理。在处理压缩、解压的过程中，是有CPU资源的消耗的。如果集群本来CPU使用率很高，就会对其他作业产生影响，不建议使用压缩。

常见的压缩格式有GZIP、BZIP2、LZO、SNAPPY。选择压缩格式要从压缩比、压缩速度考虑。不同的压缩格式对应不同的codec。

hadoop 哪个压缩编码好（BZip2Codec压缩Reduce端压缩控制）(2)

BZip2Codec压缩

package com.bigdata.compression;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.File;import java.io.FileInputStream;import java.io.FileOutputStream;public class CompressionApp { public static void main(String[] args) throws Exception{ String filename = "ip.txt"; String method = "org.apache.hadoop.io.compress.BZip2Codec"; compress(filename method); } private static void compress(String filename String method) throws Exception{ FileInputStream fis = new FileInputStream(new File(filename)); Class<?> codecClass = Class.forName(method); CompressionCodec codec = (CompressionCodec) ReflectionUtils.newInstance(codecClass new Configuration()); FileOutputStream fos = new FileOutputStream(new File(filename codec.getDefaultExtension())); CompressionOutputStream cos = codec.createOutputStream(fos); IOUtils.copyBytes(fis cos 1024*102485); cos.close(); fos.close(); fis.close(); }}

读取输入流
通过类名反射出对应的codec
写出输出流
通过IOUtils.copyBytes写出去

压缩前后对比：

hadoop 哪个压缩编码好（BZip2Codec压缩Reduce端压缩控制）(3)

hadoop 哪个压缩编码好（BZip2Codec压缩Reduce端压缩控制）(4)

BZip2Codec解压

private static void decompression(String filename) throws Exception{ CompressionCodecFactory factory = new CompressionCodecFactory(new Configuration()); CompressionCodec codec = factory.getCodec(new Path(filename)); CompressionInputStream fis = codec.createInputStream(new FileInputStream(new File(filename))); FileOutputStream fos = new FileOutputStream(new File(filename) ".decoded"); IOUtils.copyBytes(fis fos 1024*102485); fos.close(); fos.close(); fis.close();}

解压前后对比：

hadoop 哪个压缩编码好（BZip2Codec压缩Reduce端压缩控制）(5)

hadoop 哪个压缩编码好（BZip2Codec压缩Reduce端压缩控制）(6)

Map端压缩控制

mapred-default.xml中有mapreduce.map.output.compress参数，控制map输出时的压缩，mapreduce.map.output.compress.codec控制压缩类型。

代码层面可以用：

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

进行设置即可。

若要在配置层面更改，core-site.xml需要加上：

<property> <name>io.compression.codecs</name> <value> org.apache.hadoop.io.compress.GzipCodec org.apache.hadoop.io.compress.DefaultCodec org.apache.hadoop.io.compress.BZip2Codec org.apache.hadoop.io.compress.SnappyCodec </value></property>

mapred-site.xml添加：

mapreduce.map.output.compress=truemapreduce.map.output.compress.codec=org.apache.hadoop.io.compress.BZip2Codemapreduce.output.fileoutputformat.compress=truemapreduce.output.fileoutputformat.compress.codec=org.apache.hadoop.io.compress.BZip2Code

配置文件修改完重启Hadoop即可。

hadoop 哪个压缩编码好（BZip2Codec压缩Reduce端压缩控制）(7)