作者 | 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=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即可。
Reduce端壓縮控制
mapred-default.xml中有mapreduce.output.fileoutputformat.compress參數，控制reduce端輸出時的壓縮，
mapreduce.output.fileoutputformat.compress.codec控制壓縮類型。
設置同上。
你還有什麼 Hadoop 整合壓縮相關知識點？歡迎評論告訴我們 ~