基于Hive的文件格式的RCFile及其應(yīng)用是怎樣的

這篇文章給大家介紹基于Hive的文件格式的RCFile及其應(yīng)用是怎樣的，內(nèi)容非常詳細，感興趣的小伙伴們可以參考借鑒，希望對大家能有所幫助。

Hadoop 作為MR 的開源實現(xiàn)，一直以動態(tài)運行解析文件格式并獲得比MPP數(shù)據(jù)庫快上幾倍的裝載速度為優(yōu)勢。不過，MPP數(shù)據(jù)庫社區(qū)也一直批評Hadoop由于文件格式并非為特定目的而建，因此序列化和反序列化的成本過高。

1、hadoop 文件格式簡介

目前 hadoop 中流行的文件格式有如下幾種：

（1）SequenceFile

SequenceFile是Hadoop API 提供的一種二進制文件，它將數(shù)據(jù)以<key,value>的形式序列化到文件中。這種二進制文件內(nèi)部使用Hadoop 的標(biāo)準的Writable 接口實現(xiàn)序列化和反序列化。它與Hadoop API中的MapFile 是互相兼容的。Hive 中的SequenceFile 繼承自Hadoop API 的SequenceFile，不過它的key為空，使用value 存放實際的值， 這樣是為了避免MR 在運行map 階段的排序過程。如果你用Java API 編寫SequenceFile，并讓Hive 讀取的話，請確保使用value字段存放數(shù)據(jù)，否則你需要自定義讀取這種SequenceFile 的InputFormat class 和OutputFormat class。

（2）RCFile

RCFile是Hive推出的一種專門面向列的數(shù)據(jù)格式。 它遵循“先按列劃分，再垂直劃分”的設(shè)計理念。當(dāng)查詢過程中，針對它并不關(guān)心的列時，它會在IO上跳過這些列。需要說明的是，RCFile在map階段從遠端拷貝仍然是拷貝整個數(shù)據(jù)塊，并且拷貝到本地目錄后RCFile并不是真正直接跳過不需要的列，并跳到需要讀取的列， 而是通過掃描每一個row group的頭部定義來實現(xiàn)的，但是在整個HDFS Block 級別的頭部并沒有定義每個列從哪個row group起始到哪個row group結(jié)束。所以在讀取所有列的情況下，RCFile的性能反而沒有SequenceFile高。

HDFS塊內(nèi)行存儲的例子

HDFS塊內(nèi)列存儲的例子

HDFS塊內(nèi)RCFile方式存儲的例子
 

（3）Avro

Avro是一種用于支持數(shù)據(jù)密集型的二進制文件格式。它的文件格式更為緊湊，若要讀取大量數(shù)據(jù)時，Avro能夠提供更好的序列化和反序列化性能。并且Avro數(shù)據(jù)文件天生是帶Schema定義的，所以它不需要開發(fā)者在API 級別實現(xiàn)自己的Writable對象。最近多個Hadoop 子項目都支持Avro 數(shù)據(jù)格式，如Pig 、Hive、Flume、Sqoop和Hcatalog。

（4）文本格式

除上面提到的3種二進制格式之外，文本格式的數(shù)據(jù)也是Hadoop中經(jīng)常碰到的。如TextFile 、XML和JSON。 文本格式除了會占用更多磁盤資源外，對它的解析開銷一般會比二進制格式高幾十倍以上，尤其是XML 和JSON，它們的解析開銷比Textfile 還要大，因此強烈不建議在生產(chǎn)系統(tǒng)中使用這些格式進行儲存。 如果需要輸出這些格式，請在客戶端做相應(yīng)的轉(zhuǎn)換操作。 文本格式經(jīng)常會用于日志收集，數(shù)據(jù)庫導(dǎo)入，Hive默認配置也是使用文本格式，而且常常容易忘了壓縮，所以請確保使用了正確的格式。另外文本格式的一個缺點是它不具備類型和模式，比如銷售金額、利潤這類數(shù)值數(shù)據(jù)或者日期時間類型的數(shù)據(jù)，如果使用文本格式保存，由于它們本身的字符串類型的長短不一，或者含有負數(shù)，導(dǎo)致MR沒有辦法排序，所以往往需要將它們預(yù)處理成含有模式的二進制格式，這又導(dǎo)致了不必要的預(yù)處理步驟的開銷和儲存資源的浪費。

（5）外部格式

Hadoop實際上支持任意文件格式，只要能夠?qū)崿F(xiàn)對應(yīng)的RecordWriter和RecordReader即可。其中數(shù)據(jù)庫格式也是會經(jīng)常儲存在Hadoop中，比如Hbase，Mysql，Cassandra，MongoDB。 這些格式一般是為了避免大量的數(shù)據(jù)移動和快速裝載的需求而用的。他們的序列化和反序列化都是由這些數(shù)據(jù)庫格式的客戶端完成，并且文件的儲存位置和數(shù)據(jù)布局(Data Layout)不由Hadoop控制，他們的文件切分也不是按HDFS的塊大?。╞locksize）進行切割。

2、為什么需要 RCFile

Facebook曾在2010 ICDE（IEEE International Conference on Data Engineering）會議上介紹了數(shù)據(jù)倉庫Hive。Hive存儲海量數(shù)據(jù)在Hadoop系統(tǒng)中，提供了一套類數(shù)據(jù)庫的數(shù)據(jù)存儲和處理機制。它采用類SQL語言對數(shù)據(jù)進行自動化管理和處理，經(jīng)過語句解析和轉(zhuǎn)換，最終生成基于Hadoop的MapReduce任務(wù)，通過執(zhí)行這些任務(wù)完成數(shù)據(jù)處理。下圖顯示了Hive數(shù)據(jù)倉庫的系統(tǒng)結(jié)構(gòu)。 
 

Facebook在數(shù)據(jù)倉庫上遇到的存儲可擴展性的挑戰(zhàn)是獨一無二的。他們在基于Hive的數(shù)據(jù)倉庫中存儲了超過300PB的數(shù)據(jù)，并且以每日新增600TB的速度增長。去年這個數(shù)據(jù)倉庫所存儲的數(shù)據(jù)量增長了3倍?？紤]到這個增長趨勢，存儲效率問題是facebook數(shù)據(jù)倉庫基礎(chǔ)設(shè)施方面目前乃至將來一段時間內(nèi)最需要關(guān)注的。facebook工程師發(fā)表的RCFile: A Fast and Spaceefficient Data Placement Structure in MapReducebased Warehouse Systems一文，介紹了一種高效的數(shù)據(jù)存儲結(jié)構(gòu)——RCFile（Record Columnar File），并將其應(yīng)用于Facebook的數(shù)據(jù)倉庫Hive中。與傳統(tǒng)數(shù)據(jù)庫的數(shù)據(jù)存儲結(jié)構(gòu)相比，RCFile更有效地滿足了基于MapReduce的數(shù)據(jù)倉庫的四個關(guān)鍵需求，即Fast data loading、Fast query processing、Highly efficient storage space utilization和Strong adaptivity to highly dynamic workload patterns。RCFile 廣泛應(yīng)用于Facebook公司的數(shù)據(jù)分析系統(tǒng)Hive中。首先，RCFile具備相當(dāng)于行存儲的數(shù)據(jù)加載速度和負載適應(yīng)能力；其次，RCFile的讀優(yōu)化可以在掃描表格時避免不必要的列讀取，測試顯示在多數(shù)情況下，它比其他結(jié)構(gòu)擁有更好的性能；再次，RCFile使用列維度的壓縮，因此能夠有效提升存儲空間利用率。
為了提高存儲空間利用率，F(xiàn)acebook各產(chǎn)品線應(yīng)用產(chǎn)生的數(shù)據(jù)從2010年起均采用RCFile結(jié)構(gòu)存儲，按行存儲（SequenceFile/TextFile）結(jié)構(gòu)保存的數(shù)據(jù)集也轉(zhuǎn)存為RCFile格式。此外，Yahoo公司也在Pig數(shù)據(jù)分析系統(tǒng)中集成了RCFile，RCFile正在用于另一個基于Hadoop的數(shù)據(jù)管理系統(tǒng)Howl（http://wiki.apache.org/pig/Howl）。而且，根據(jù)Hive開發(fā)社區(qū)的交流，RCFile也成功整合加入其他基于MapReduce的數(shù)據(jù)分析平臺。有理由相信，作為數(shù)據(jù)存儲標(biāo)準的RCFile，將繼續(xù)在MapReduce環(huán)境下的大規(guī)模數(shù)據(jù)分析中扮演重要角色。
 

3、RCFile 簡介

facebook 的數(shù)據(jù)倉庫中數(shù)據(jù)被加載到表里面時首先使用的存儲格式是Facebook自己開發(fā)的Record-Columnar File Format(RCFile)。RCFile是一種“允許按行查詢，提供了列存儲的壓縮效率”的混合列存儲格式。它的核心思想是首先把Hive表水平切分成多個行組（row groups），然后組內(nèi)按照列垂直切分，這樣列與列的數(shù)據(jù)在磁盤上就是連續(xù)的存儲塊了。
當(dāng)一個行組內(nèi)的所有列寫到磁盤時，RCFile就會以列為單位對數(shù)據(jù)使用類似zlib/lzo的算法進行壓縮。當(dāng)讀取列數(shù)據(jù)的時候使用惰性解壓策略（ lazy decompression），也就是說用戶的某個查詢?nèi)绻皇巧婕暗揭粋€表中的部分列的時候，RCFile會跳過不需要的列的解壓縮和反序列化的過程。通過在facebook的數(shù)據(jù)倉庫中選取有代表性的例子實驗，RCFile能夠提供5倍的壓縮比。

 

4、超越RCFile，下一步采用什么方法

隨著數(shù)據(jù)倉庫中存儲的數(shù)據(jù)量持續(xù)增長，F(xiàn)B組內(nèi)的工程師開始研究提高壓縮效率的技術(shù)和方法。研究的焦點集中在列級別的編碼方法，例如行程長度編碼（run-length encoding）、詞典編碼（dictionary encoding）、參考幀編碼（frame of reference encoding）、能夠在通用壓縮過程之前更好的在列級別降低邏輯冗余的數(shù)值編碼方法。FB也嘗試過新的列類型（例如JSON是在Facebook內(nèi)部廣泛使用的格式，把JSON格式的數(shù)據(jù)按照結(jié)構(gòu)化的方式存儲既可以滿足高效查詢的需求，同時也降低了JSON元數(shù)據(jù)存儲的冗余）。FB的實驗表明列級別的編碼如果使用得當(dāng)?shù)脑捘軌蝻@著提高RCFile的壓縮比。
與此同時，Hortonworks也在嘗試類似的思路去改進Hive的存儲格式。Hortonworks的工程團隊設(shè)計和實現(xiàn)了ORCFile（包括存儲格式和讀寫接口），這幫助Facebook的數(shù)據(jù)倉庫設(shè)計和實現(xiàn)新的存儲格式提供了一個很好的開始。
 

關(guān)于 ORCFile 的介紹請見這里：http://yanbohappy.sinaapp.com/?p=478

關(guān)于性能評測，筆者這里暫時沒有條件，貼一張某次 hive 技術(shù)峰會演講嘉賓的截圖：

5、如何生成 RCFile 文件

上面說了這么多，想必你已經(jīng)知道 RCFile 主要用于提升 hive 的查詢效率，那如何生成這種格式的文件呢？

（1）hive 中直接通過textfile表進行insert轉(zhuǎn)換

例如：

insert overwrite table http_RCTable partition(dt='2013-09-30') select p_id,tm,idate,phone from tmp_testp where dt='2013-09-30';

（2）通過 mapreduce 生成

目前為止，mapreduce 并沒有提供內(nèi)置 API 對 RCFile 進行支持，倒是 pig、hive、hcatalog 等 hadoop生態(tài)圈里的其他項目進行了支持，究其原因是因為 RCFile 相比 textfile 等其它文件格式，對于 mapreduce 的應(yīng)用場景來說沒有顯著的優(yōu)勢。
 

為了避免重復(fù)造輪子，下面的生成 RCFile 的 mapreduce 代碼調(diào)用了 hive 和 hcatalog 的相關(guān)類，注意你在測試下面的代碼時，你的 hadoop、hive、hcatalog 版本要一致，否則。。。你懂的。。。

比如我用的 hive-0.10.0+198-1.cdh5.4.0，那么就應(yīng)該下載對應(yīng)的版本：http://archive.cloudera.com/cdh5/cdh/4/

PS：下面的代碼已經(jīng)測試通過，木有問題。

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.serde2.columnar.BytesRefArrayWritable;
import org.apache.hadoop.hive.serde2.columnar.BytesRefWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.hcatalog.rcfile.RCFileMapReduceInputFormat;
import org.apache.hcatalog.rcfile.RCFileMapReduceOutputFormat;



public class TextToRCFile extends Configured implements Tool{

	
	
	public static class Map 
    	extends Mapper<Object, Text, NullWritable, BytesRefArrayWritable>{
		
		private byte[] fieldData;
		private int numCols;
		private BytesRefArrayWritable bytes;
		
		@Override
		protected void setup(Context context) throws IOException, InterruptedException {
			numCols = context.getConfiguration().getInt("hive.io.rcfile.column.number.conf", 0);
			bytes = new BytesRefArrayWritable(numCols);
		}
		
		public void map(Object key, Text line, Context context
                ) throws IOException, InterruptedException {
			bytes.clear();
			String[] cols = line.toString().split("\\|");
			System.out.println("SIZE : "+cols.length);
			for (int i=0; i<numCols; i++){
	        	fieldData = cols[i].getBytes("UTF-8");
	        	BytesRefWritable cu = null;
	            cu = new BytesRefWritable(fieldData, 0, fieldData.length);
	            bytes.set(i, cu);
	        }
			context.write(NullWritable.get(), bytes);
		}
	}
	
	@Override
	public int run(String[] args) throws Exception {
		Configuration conf = new Configuration();
		String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
		if(otherArgs.length < 2){
	    	System.out.println("Usage: " +
	    			"hadoop jar RCFileLoader.jar <main class> " +
	    			"-tableName <tableName> -numCols <numberOfColumns> -input <input path> " +
	    			"-output <output path> -rowGroupSize <rowGroupSize> -ioBufferSize <ioBufferSize>");
	    	System.out.println("For test");
	    	System.out.println("$HADOOP jar RCFileLoader.jar edu.osu.cse.rsam.rcfile.mapreduce.LoadTable " +
	    			"-tableName test1 -numCols 10 -input RCFileLoaderTest/test1 " +
	    			"-output RCFileLoaderTest/RCFile_test1");
	    	System.out.println("$HADOOP jar RCFileLoader.jar edu.osu.cse.rsam.rcfile.mapreduce.LoadTable " +
	    			"-tableName test2 -numCols 5 -input RCFileLoaderTest/test2 " +
	    			"-output RCFileLoaderTest/RCFile_test2");
	    	return 2;
	    }
		
		/* For test
		   
		 */
		
	    
		String tableName = "";
		int numCols = 0;
		String inputPath = "";
		String outputPath = "";
		int rowGroupSize = 16 *1024*1024;
		int ioBufferSize = 128*1024;
	    for (int i=0; i<otherArgs.length - 1; i++){
	    	if("-tableName".equals(otherArgs[i])){
	    		tableName = otherArgs[i+1];
	    	}else if ("-numCols".equals(otherArgs[i])){
	    		numCols = Integer.parseInt(otherArgs[i+1]);
	    	}else if ("-input".equals(otherArgs[i])){
	    		inputPath = otherArgs[i+1];
	    	}else if("-output".equals(otherArgs[i])){
	    		outputPath = otherArgs[i+1];
	    	}else if("-rowGroupSize".equals(otherArgs[i])){
	    		rowGroupSize = Integer.parseInt(otherArgs[i+1]);
	    	}else if("-ioBufferSize".equals(otherArgs[i])){
	    		ioBufferSize = Integer.parseInt(otherArgs[i+1]);
	    	}
	    	
	    }
	    
	    conf.setInt("hive.io.rcfile.record.buffer.size", rowGroupSize);
	    conf.setInt("io.file.buffer.size", ioBufferSize);
	    
	    Job job = new Job(conf, "RCFile loader: loading table " + tableName + " with " + numCols + " columns");
	    
	    job.setJarByClass(TextToRCFile.class);
	    job.setMapperClass(Map.class);
	    job.setMapOutputKeyClass(NullWritable.class);
	    job.setMapOutputValueClass(BytesRefArrayWritable.class);
//	    job.setNumReduceTasks(0);
	    
	    FileInputFormat.addInputPath(job, new Path(inputPath));
	    
	    job.setOutputFormatClass(RCFileMapReduceOutputFormat.class);
	    RCFileMapReduceOutputFormat.setColumnNumber(job.getConfiguration(), numCols);
	    RCFileMapReduceOutputFormat.setOutputPath(job, new Path(outputPath));
	    RCFileMapReduceOutputFormat.setCompressOutput(job, false);
	    
	    
	    System.out.println("Loading table " + tableName + " from " + inputPath + " to RCFile located at " + outputPath);
	    System.out.println("number of columns:" + job.getConfiguration().get("hive.io.rcfile.column.number.conf"));
	    System.out.println("RCFile row group size:" + job.getConfiguration().get("hive.io.rcfile.record.buffer.size"));
	    System.out.println("io bufer size:" + job.getConfiguration().get("io.file.buffer.size"));
	    
	    return (job.waitForCompletion(true) ? 0 : 1);
	}
	
	public static void main(String[] args) throws Exception {
	    int res = ToolRunner.run(new Configuration(), new TextToRCFile(), args);
	    System.exit(res);
	}

}

關(guān)于基于Hive的文件格式的RCFile及其應(yīng)用是怎樣的就分享到這里了，希望以上內(nèi)容可以對大家有一定的幫助，可以學(xué)到更多知識。如果覺得文章不錯，可以把它分享出去讓更多的人看到。