作者 | 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。
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写出去
压缩前后对比：
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();}
解压前后对比：
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=true
mapreduce.map.output.compress.codec=org.apache.hadoop.io.compress.BZip2Codemapreduce.output.fileoutputformat.compress=true
mapreduce.output.fileoutputformat.compress.codec=org.apache.hadoop.io.compress.BZip2Code
配置文件修改完重启Hadoop即可。
Reduce端压缩控制
mapred-default.xml中有mapreduce.output.fileoutputformat.compress参数，控制reduce端输出时的压缩，
mapreduce.output.fileoutputformat.compress.codec控制压缩类型。
设置同上。
你还有什么 Hadoop 整合压缩相关知识点？欢迎评论告诉我们 ~
《原力计划【第二季】- 学习力挑战》正式开始！即日起至 3月21日，千万流量支持原创作者！更有专属【勋章】等你来挑战
推荐阅读：6 个步骤，教你在Ubuntu虚拟机环境下，用Docker自带的DNS配置Hadoop | 附代码
删库跑路事件发生，SaaS云服务如何守护数据安全
释放低代码小宇宙，微软 Power Platform 震撼来袭！
闪电网络的 5 个优点和4 个缺点、本质、来源与工作原理……一文带你读懂闪电网络！
乔布斯遗孀裸捐 250 亿美元财产：没兴趣累积财富
赔偿谷歌1.8亿美元！前Uber自动驾驶主管被告到破产
真香，朕在看了！
