作者 |?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自動駕駛主管被告到破產 真香，朕在看了！

總結

以上是生活随笔為你收集整理的BZip2Codec压缩、Map端压缩控制、Reduce端压缩控制……都在这份Hadoop整合压缩知识点里了！...的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。