Hadoop有一些數(shù)據(jù)I/O方面操作的工具，其中一些比Hadoop使用的都更普遍。例如數(shù)據(jù)完整性和壓縮。但是當(dāng)使用這些工具處理多達(dá)幾TB數(shù)據(jù)的時候，仍然需要特別注意。其它的一些工具或APIs構(gòu)建成了模塊用于開發(fā)分布式系統(tǒng)，例如:序列化框架和硬盤上的數(shù)據(jù)結(jié)構(gòu)化。

數(shù)據(jù)完整性

Hadoop的使用者肯定都希望數(shù)據(jù)在存儲或處理的過程中不會丟失或損壞。然而，由于硬盤或網(wǎng)絡(luò)的每一次I/O操作都有可能在讀或?qū)懙倪^程中出錯，Hadoop經(jīng)常處理的數(shù)據(jù)量都相當(dāng)大，當(dāng)這樣地大量數(shù)據(jù)在系統(tǒng)中傳輸時，數(shù)據(jù)損壞的概率還是挺高的。

通常用于判斷數(shù)據(jù)是否損壞所使用的方法是當(dāng)數(shù)據(jù)第一次進(jìn)入系統(tǒng)時候計算一個校驗和，并不管在任何時候，數(shù)據(jù)被傳輸通過一個不可信的通道后，再一次計算校驗和，就可以判斷數(shù)據(jù)是否損壞。如果新生成的校驗和與原始的校驗和不一樣，就認(rèn)為數(shù)據(jù)遭到了損壞。這個技術(shù)不能夠修復(fù)數(shù)據(jù)，僅僅是檢查數(shù)據(jù)是否損壞。(這也是不使用低檔的硬件的原因，特殊情況下，肯定要使用ECC內(nèi)存)。注意：有時校驗和也可能損壞，而數(shù)據(jù)沒有損壞，但這種可能性很低，因為校驗和比數(shù)據(jù)小多了。

通常使用的錯誤檢測碼是CRC-32(32位循環(huán)冗余校驗)，不管輸入的數(shù)據(jù)量多大，都計算出一個32的整數(shù)校驗和。CRC-32在Hadoop的ChecksumFileSystem類中用于計算校驗和。然而HDFS使用的是另一種更有效的方式，叫做CRC-32C。

HDFS中的數(shù)據(jù)完整性

HDFS透明地計算寫入的所有數(shù)據(jù)的校驗和，并且默認(rèn)情況下會在讀取數(shù)據(jù)時驗證校驗和。每個用dfs.bytes-per-checksum屬性指定大小的數(shù)據(jù)塊都會單獨(dú)生成一個校驗和。默認(rèn)是512字節(jié)。因為CRC-32C校驗和是4個字節(jié)，所以存儲開銷少于1%。((12810241024)/5124/12810241024)100%=0.78%。

數(shù)據(jù)節(jié)點(diǎn)有責(zé)任在接收到數(shù)據(jù)后存儲數(shù)據(jù)前校驗數(shù)據(jù)，并計算校驗和，不管從客戶端接收數(shù)據(jù)還是在復(fù)制復(fù)本數(shù)據(jù)期間從其它節(jié)點(diǎn)接收數(shù)據(jù)。正在寫入數(shù)據(jù)到HDFS的客戶端將校驗和傳入數(shù)據(jù)節(jié)點(diǎn)的管道中(如第3章介紹的那樣)，管道中的最后一個數(shù)據(jù)節(jié)點(diǎn)驗證校驗和。如果數(shù)據(jù)節(jié)點(diǎn)檢測到錯誤，客戶端會收到一個IOExcption子類的異常，客戶端應(yīng)該采取應(yīng)用程序特定的措施，比如，重新嘗試寫入。

當(dāng)客戶端從數(shù)據(jù)節(jié)點(diǎn)讀取數(shù)據(jù)的時候，它們也要驗證校驗和，將讀取的數(shù)據(jù)計算出的校驗和和存儲在數(shù)據(jù)節(jié)點(diǎn)中的校驗和進(jìn)行比較。每一個數(shù)據(jù)節(jié)點(diǎn)都永久保存了一份驗證校驗和的歷史日志。所以HDFS知道每一個數(shù)據(jù)塊最近驗證校驗和的時間。當(dāng)客戶端成功地校驗了某一塊數(shù)據(jù)的時候，它會告訴數(shù)據(jù)節(jié)點(diǎn)，數(shù)據(jù)節(jié)點(diǎn)會更新日志。保存這些校驗的信息可以用于檢測壞的硬盤。

除了客戶端讀取數(shù)據(jù)時的塊校驗，每一個數(shù)據(jù)節(jié)點(diǎn)都會以一個后臺線程方式執(zhí)行一個DataBlockScanner,周期性檢測存儲在這個數(shù)據(jù)節(jié)點(diǎn)上的數(shù)據(jù)塊是否損壞。這是為了防止由于物理存儲媒介的"位衰減"造成的數(shù)據(jù)損壞?？梢钥吹?1章的"數(shù)據(jù)節(jié)點(diǎn)塊掃描"相關(guān)內(nèi)容詳細(xì)了解如何獲取掃描的報告。

由于HDFS存儲著數(shù)據(jù)塊的復(fù)本，所以它能夠通過復(fù)制一個好的復(fù)本生成一個新的，沒有損壞的復(fù)本來修復(fù)已經(jīng)損壞的塊。具體實現(xiàn)方法是客戶端在讀取塊數(shù)據(jù)時如果檢測到一個錯誤，它會向名稱節(jié)點(diǎn)報告這個壞的塊和塊所在的數(shù)據(jù)節(jié)點(diǎn)，然后拋出一個ChecksumException異常。名稱節(jié)點(diǎn)會將這個塊狀態(tài)標(biāo)記為"損壞"，并不再將其它客戶端提供這個塊的地址，然后在另外一個數(shù)據(jù)節(jié)點(diǎn)中生成這個塊的一個復(fù)本，直到達(dá)到設(shè)置的塊復(fù)本數(shù)，一旦完成后，這個損壞的塊就會被刪除。

我們可以在調(diào)用FileSystem的open方法打開文件之前，調(diào)用它的setVerifyChecksum方法，傳遞一個false值，就可以關(guān)閉校驗和的驗證。如果使用shell命令，我們可以在-get或-copyToLocal命令中添加-ignoreCrc選項同樣可以關(guān)閉校驗和的驗證。如果你想要看看這部分損壞的數(shù)據(jù)，再進(jìn)行相應(yīng)處理的話，這個功能將是有用的。例如，你也許想在刪除損壞的塊數(shù)據(jù)之前想要看看是否能夠恢復(fù)塊中的數(shù)據(jù)。

你可以通過使用hadoop fs -checksum來得到一個文件的校驗和。這對于檢查HDFS中兩個文件是否有相同的內(nèi)容是有用的(distcp命令也可以做這些事情)，可以參看"使用distcp命令并發(fā)復(fù)制"小節(jié)舉的示例。

LocalFileSystem

Hadoop LocalFileSystem在客戶端驗證校驗和。這意味著當(dāng)你向一個叫"filename"的文件中寫入數(shù)據(jù)時，文件系統(tǒng)客戶端會在文件相同目錄透明的創(chuàng)建一個隱藏的文件，叫做 .filename.crc，這個文件包含每一段文件數(shù)據(jù)的檢驗和。段的大小由屬性file.bytes-per-checksum定義，默認(rèn)是512字節(jié)。段大小做為元數(shù)據(jù)存儲在.crc后綴的文件中，所以即使段大小的設(shè)置以后改變了，文件仍然可以完好無損地讀取。當(dāng)讀取文件時，會驗證檢驗和，如果出現(xiàn)錯誤，LocalFileSystem拋出ChecksumException異常。

計算校驗和的代價很小(JAVA中，使用native代碼計算)，不會對讀取或?qū)懭胛募斐墒裁从绊?。對于大多?shù)應(yīng)用來說，為了保證數(shù)據(jù)的完整性這是完全可以承受的代價。然而，如果底層的文件系統(tǒng)本身支持校驗，就可以關(guān)閉LocalFileSystem的校驗。此時可以使用RawLocalFileSysem代替LocalFileSystem。要想在應(yīng)用的全局范圍內(nèi)關(guān)閉校驗和，可以將屬性fs.file.impl的值設(shè)置為org.apache.hadoop.fs.RawLocalFileSystem。如果你僅僅想對某一些讀取操作關(guān)閉檢驗和驗證時，也可以直接創(chuàng)建一個RawLocalFileSystem實例。例如:

Configuration conf = ...
FileSystem fs = new RawLocalFileSystem();
fs.initialize(null, conf);

ChecksumFileSystem

LocalFileSystem使用CheckFileSystem來實現(xiàn)校驗和的功能。這個類可以很方便地在其它沒有校驗和功能的文件系統(tǒng)中加入校驗和功能。ChecksumFileSystem僅僅是FileSystem的一個封裝類。一般的寫法如下:

FileSystem rawFs = ...
FileSystem checksummedFs = new ChecksumFileSystem(rawFs);

底層的文件系統(tǒng)叫做原生文件系統(tǒng)，可以通過ChecksumFileSystem的getRawFileSystem方法獲得。ChecksumFileSystem還有一些和校驗和相關(guān)的有用的方法，例如getChecksumFile()方法用于獲取任意文件的校驗和文件的路徑。可以查看相關(guān)參考資料看看其它的方法。

如果在讀取文件的過程中，ChecksumFileSystem檢測到一個錯誤。它將會調(diào)用它的reportChecksumFailure()方法。這個方法默認(rèn)的實現(xiàn)沒有做任何事情。但是LocalFileSystem將檢驗失敗的文件和它的校驗和單獨(dú)移到同一設(shè)備中叫做"bad_files"的目錄中。管理員應(yīng)該定期地檢查這個目錄中是否有壞的文件并處理。

壓縮

文件壓縮有兩個主要的好處：減少存儲文件所需的空間;加快數(shù)據(jù)在網(wǎng)絡(luò)傳輸?shù)乃俾屎蛷挠脖P讀寫文件的速度。當(dāng)處理海量數(shù)據(jù)時，這些好處帶來的效果是很可觀的，所以它值得我們仔細(xì)考慮如何在Hadoop使用壓縮。

有多種不同的壓縮格式，工具和算法。每一種都有不同的特性。表5-1例舉出了Hadoop常用的一些壓縮工具/格式。

所有的壓縮算法都會在空間與時間兩方面進(jìn)行權(quán)衡:更快地壓縮與解壓縮算法通常以節(jié)約更小的空間為代價。表5-1所列的壓縮工具一般都會提供9個不同的選擇來在空間與時間上進(jìn)行權(quán)衡。-1意思是優(yōu)化速度，-9意思是優(yōu)化空間。例如下面的命令以最快的壓縮速度創(chuàng)建了一個叫做file.gz的壓縮文件。

% gzip -1 file

不同的壓縮工具有著非常不同的壓縮特性。通常選擇gzip壓縮格式，所有壓縮工具中，gzip在空間與時間權(quán)衡方面性能居中。bzip2比gzip有更高的壓縮率，但是更慢。bzip2的解壓速度比它的壓縮速度快，但是還是比其它壓縮格式要慢。另一方面，LZO、LZ4和Snappy都優(yōu)化了速度，在速度上按照順序依次更快，且都比gzip快很多。但是壓縮率不如gzip高。Snappy和LZ4比LZO的解壓速度要快很多^[3]。

表5-1的"可切片"一列表示壓縮格式是否支持切片(也就是說你可以定位到數(shù)據(jù)流的任意一點(diǎn)，并從這一點(diǎn)開始讀取數(shù)據(jù))?？汕衅膲嚎s格式特別適合MapReduce?？梢钥瓷院蟮?壓縮與切片"小節(jié)進(jìn)一步討論。

編解碼器

編解碼器是壓縮-解壓算法的實現(xiàn)。Hadoop中，一個編解碼器是CompressionCodec接口的一個實現(xiàn)類。例如:GzipCodec封裝了gzip的壓縮與解壓算法。表5-2列舉出了Hadoop中的編解碼器。

LZO庫是遵循GPL協(xié)議的，所以在Apache發(fā)布的Hadoop中也許沒有包含，所以需要單獨(dú)從Google(或GitHub,包含bug修復(fù)和更多工具)下載。
LzopCodec是一個兼容lzop工具的實現(xiàn)類，本質(zhì)上，是帶有額外頭文件的LZO壓縮工具類。也有一個純LZO壓縮格式的實現(xiàn)類叫做LzoCodec,它使用的是.lzo_deflate文件擴(kuò)展名(就如同DEFLATE，是一個沒有頭文件的gzip)。

使用CompressionCodec壓縮與解壓流
CompressionCodec有兩個方法，可以讓你很容易的壓縮或解壓數(shù)據(jù)。為了在寫入輸出流時壓縮數(shù)據(jù)，你可以使用createOutputStream(OutputStream out)方法創(chuàng)建一個CompressionOutputStream輸出流，然后你將未壓縮的數(shù)據(jù)寫入，CompressionOutputStream就可以讓數(shù)據(jù)以壓縮的形式寫入底層輸出流中。相應(yīng)地，要想從一個輸入流中解壓數(shù)據(jù)，調(diào)用createInputStream(InpuStream in)方法獲取一個CompressionInputStream對象，這個對象可以讓你從底層流的壓縮數(shù)據(jù)中讀取出解壓后的數(shù)據(jù)。

CompressionOutputStream和CompressionInputStream與java.util.zip.DeflaterOutputStream和java.util.zip.DeflaterInputStream相似。不過，前兩者比后兩者多個功能，可以重置底層的壓縮或解壓器。這對于想分塊壓縮數(shù)據(jù)的應(yīng)用來說是很重要的。例如本章后面將講解的SequenceFile。

示例5-1展示了如何將從標(biāo)準(zhǔn)輸入流讀取的數(shù)據(jù)壓縮，然后寫入標(biāo)準(zhǔn)輸出流中。

示例5-1:壓縮讀取的數(shù)據(jù)并寫入標(biāo)準(zhǔn)輸出流的程序
public class StreamCompressor {
   public static void main(String[] args) throws Exception {
      String codecClassname = args[0];
     Class<?> codecClass = Class.forName(codecClassname);
     Configuration conf = new Configuration();
     CompressionCodec codec = (CompressionCodec)
    ReflectionUtils.newInstance(codecClass, conf);
    CompressionOutputStream out =      codec.createOutputStream(System.out);
    IOUtils.copyBytes(System.in, out, 4096, false);
    out.finish();
  }
}

這個應(yīng)用需要一個 CompressionCodec實現(xiàn)類的全類名做為命令行第一個參數(shù)。我們使用ReflectionUtils創(chuàng)建一個編碼類的實例，然后獲取一個System.out的壓縮封裝類，然后我們調(diào)用IOUtils的copyBytes方法將輸入流中的數(shù)據(jù)復(fù)制到輸出流中。在輸出之前，數(shù)據(jù)會經(jīng)過 CompressionOutputStream壓縮。最后調(diào)用 CompressionOutputStream的finish方法告訴壓縮器停止向輸出流中寫入壓縮的數(shù)據(jù)，但不關(guān)閉輸出流。我們可以用下面的命令進(jìn)行嘗試，壓縮"Text"字符串，調(diào)用StreamCompressor類，傳入 GzipCodec壓縮器，然后用gunzip進(jìn)行解壓。

% echo "Text" | hadoop StreamCompressor org.apache.hadoop.io.compress.GzipCodec \
| gunzip -
Text

使用CompressionCodecFactory推斷出壓縮編碼
如果你需要讀取一個壓縮文件，正常情況下你會觀察壓縮文件的擴(kuò)展名來推斷出所使用的壓縮編碼，.gz結(jié)尾的文件可以使用 GzipCodec讀取等等。每一種壓縮編碼的擴(kuò)展名都在表5-1中列舉。CompressionCodecFactory通過它的getCodec()方法提供了一種通過擴(kuò)展名得到相應(yīng)壓縮編碼的方法。示例5-2顯示了如何使用這種方法解壓文件。

示例:5-2 從文件擴(kuò)展名推斷出解壓縮編碼來對文件進(jìn)行解壓的程序
public class FileDecompressor {
public static void main(String[] args) throws Exception {
  String uri = args[0];
  Configuration conf = new Configuration();
  FileSystem fs = FileSystem.get(URI.create(uri), conf);
  Path inputPath = new Path(uri);
  CompressionCodecFactory factory = new CompressionCodecFactory(conf);
  CompressionCodec codec = factory.getCodec(inputPath);
  if (codec == null) {
   System.err.println("No codec found for " + uri);
  System.exit(1);
  }
  String outputUri =
  CompressionCodecFactory.removeSuffix(uri, codec.getDefaultExtension());
  InputStream in = null;
  OutputStream out = null;
  try {
   in = codec.createInputStream(fs.open(inputPath));
   out = fs.create(new Path(outputUri));
   IOUtils.copyBytes(in, out, conf);
  } finally {
    IOUtils.closeStream(in);
    IOUtils.closeStream(out);
    }
  }
}

一旦推斷出了編碼，就會使用 CompressionCodecFactory的 removeSuffix() 靜態(tài)方法去掉文件的后綴得到輸出文件名。通過這種方法，使用如下命令，一個叫做file.gz的文件會被解壓成file。

% hadoop FileDecompressor file.gz

CompressionCodecFactory會加載表5-2中除了LZO之外的所有編碼，和表5-3中屬性 io.compression.codecs所列舉的所有編碼，默認(rèn)情況下，這個屬性值為空，如果你希望注冊一個自定義的編碼(例如這個外部的LZO編碼)，你就需要修改這個屬性。每一種編碼都知道它自己的默認(rèn)文件擴(kuò)展名，因此CompressionCodecFactory就可以通過搜索所有注冊的編碼找到與給定文件名匹配的編碼。

表5-3:壓縮編碼配置屬性

本地庫
從性能考慮，推薦使用本地庫來進(jìn)行壓縮與解壓縮。例如一項測試表明，使用本地庫中的gzip比使用java內(nèi)建的gzip實現(xiàn)至多減少50%解壓縮時間和大約10%的壓縮時間。表5-4顯示了對于所有壓縮格式，java與本地庫是否提供了實現(xiàn)。所有的格式都有本地實現(xiàn)，但并不是都有java實現(xiàn)，例如LZO壓縮格式。

表5-4:壓縮庫實現(xiàn)

Apache Hadoop的二進(jìn)制tar包帶有預(yù)建的針對64位Linux系統(tǒng)的壓縮編碼本地庫實現(xiàn)，叫做 libhadoop.so。對于其它平臺，你需要按照Hadoop源碼頂級目錄中BUILDING.txt文件說明的那樣編譯自己的庫文件。

通過設(shè)置java系統(tǒng)屬性 java.library.path可以找到本地庫，可以在hadoop的etc/hadoop目錄中進(jìn)行設(shè)置。如果你不想在hadoop配置文件中設(shè)置，你就需要在應(yīng)用程序中設(shè)置。

默認(rèn)情況下，Hadoop會自動尋找與它運(yùn)行平臺相對應(yīng)的本地庫，如果發(fā)現(xiàn)了，就會自動加載。這就意味著你不需要為了使用本地庫而做任何配置。然后，在某些情況下，你希望不使用本地庫，例如你想調(diào)度一個壓縮相關(guān)的問題時。你可以通過將屬性io.native.lib.available設(shè)置成false來不使用本地庫，而使用Java內(nèi)建的編碼(如果可以找得到的話)。

編碼池
如果你正在使用本地庫，而且在你的應(yīng)用程序中需要做大量的壓縮與解壓縮工作，可以考慮使用編碼池(CodecPool)。編碼池可以讓你重用解壓縮對象，而省去了創(chuàng)建這些對象的開銷。

示例5-3中的代碼顯示了編碼池的應(yīng)用。雖然本例實際上只需要創(chuàng)建一個解壓縮對象，沒有必要使用編碼池。

示例5-3:從標(biāo)準(zhǔn)輸入中讀取數(shù)據(jù)，使用編碼池壓縮器將數(shù)據(jù)寫入標(biāo)準(zhǔn)輸出中
public class PooledStreamCompressor {
  public static void main(String[] args) throws Exception {
    String codecClassname = args[0];
   Class<?> codecClass = Class.forName(codecClassname);
   Configuration conf = new Configuration();
  CompressionCodec codec = (CompressionCodec)
   ReflectionUtils.newInstance(codecClass, conf);
   Compressor compressor = null;
   try {
    compressor = CodecPool.getCompressor(codec);
    CompressionOutputStream out =
      codec.createOutputStream(System.out, compressor);
      IOUtils.copyBytes(System.in, out, 4096, false);
     out.finish();
  } finally {
     CodecPool.returnCompressor(compressor);
  }
}
}

對于給定的 CompressionCodec，我們可以從編碼池中獲取壓縮實例。然后在CompressionCodec的 createOutputStream() 方法中使用這個壓縮實例。通過使用finally塊，我們就可以確保，這個壓縮實例在使用完以后能夠回到編碼池中，即使在從標(biāo)準(zhǔn)輸入流復(fù)制數(shù)據(jù)到標(biāo)準(zhǔn)輸出流中時，出現(xiàn)了IOException異常。

未完待續(xù).........

1. DEFLATE壓縮算法標(biāo)準(zhǔn)實現(xiàn)是zlib.但是沒有生成DEFLATE格式的通用命令行工具。通常使用gzip工具(gzip使用DEFLATE壓縮算法，不過增加了額外的頭和尾)。.deflate格式的擴(kuò)展文件是Hadoop使用的。 ?

2. 如果在預(yù)處理過程中，已經(jīng)創(chuàng)建了索引，LZO壓縮的文件可以切片，見后面的壓縮的切片小節(jié)介紹。 ?

3. 想了解更多綜合性壓縮標(biāo)準(zhǔn)，可以參看jvm壓縮標(biāo)準(zhǔn)了解JVM兼容的一些庫(包庫一些native庫) ?