devbox
小蓝xlanll
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

大数据系统-Hive的3种数据存储格式_hive元数据存储的三种模式

作者：小蓝xlanll | 2024-05-03 10:48:09

赞

踩

hive元数据存储的三种模式

    关系数据库里有表（table），分区，hive里也有这些东西，这些东西在hive技术里称为hive的数据模型。今天本文介绍hive的数据类型，数据模型以及文件存储格式。这些知识大家可以类比关系数据库的相关知识。

　　首先我要讲讲hive的数据类型。

　　Hive支持两种数据类型，一类叫原子数据类型，一类叫复杂数据类型。

　　原子数据类型包括数值型、布尔型和字符串类型，具体如下表所示：

基本数据类型

类型

描述

示例

TINYINT

1个字节（8位）有符号整数

1

SMALLINT

2字节（16位）有符号整数

1

INT

4字节（32位）有符号整数

1

BIGINT

8字节（64位）有符号整数

1

FLOAT

4字节（32位）单精度浮点数

1.0

DOUBLE

8字节（64位）双精度浮点数

1.0

BOOLEAN

true/false

true

STRING

字符串

‘xia’,”xia”

 

　　由上表我们看到hive不支持日期类型，在hive里日期都是用字符串来表示的，而常用的日期格式转化操作则是通过自定义函数进行操作。

　　hive是用java开发的，hive里的基本数据类型和java的基本数据类型也是一一对应的，除了string类型。有符号的整数类型：TINYINT、SMALLINT、INT和BIGINT分别等价于java的byte、short、int和long原子类型，它们分别为1字节、2字节、4字节和8字节有符号整数。Hive的浮点数据类型FLOAT和DOUBLE,对应于java的基本类型float和double类型。而hive的BOOLEAN类型相当于java的基本数据类型boolean。

　　对于hive的String类型相当于数据库的varchar类型，该类型是一个可变的字符串，不过它不能声明其中最多能存储多少个字符，理论上它可以存储2GB的字符数。

　　Hive支持基本类型的转换，低字节的基本类型可以转化为高字节的类型，例如TINYINT、SMALLINT、INT可以转化为FLOAT，而所有的整数类型、FLOAT以及STRING类型可以转化为DOUBLE类型，这些转化可以从java语言的类型转化考虑，因为hive就是用java编写的。当然也支持高字节类型转化为低字节类型，这就需要使用hive的自定义函数CAST了。

　　复杂数据类型包括数组（ARRAY）、映射（MAP）和结构体（STRUCT），具体如下表所示：

复杂数据类型

类型

描述

示例

ARRAY

一组有序字段。字段的类型必须相同

Array(1,2)

MAP

一组无序的键/值对。键的类型必须是原子的，值可以是任何类型，同一个映射的键的类型必须相同，值得类型也必须相同

Map(‘a’,1,’b’,2)

STRUCT

一组命名的字段。字段类型可以不同

Struct(‘a’,1,1,0)

 

下面我们看看hive使用复杂数据类型的实例，建表：

Create  table  complex(col1 ARRAY< INT >,
Col2 MAP<STRING, INT >,
Col3 STRUCT<a:STRING,b : INT ,c: DOUBLE >);

　　

查询语句：

Select  col1[0],col2[‘b’],col3.c from  complex;

　　

　　接下来我们来看看hive的数据模型，hive的数据模型包括：database、table、partition和bucket。下面我将一一论述这四种数据模型。

1.Database：相当于关系数据库里的命名空间（namespace），它的作用是将用户和数据库的应用隔离到不同的数据库或模式中，该模型在hive 0.6.0之后的版本支持，hive提供了create database dbname、use dbname以及drop database dbname这样的语句。

2.表（table）：hive的表逻辑上由存储的数据和描述表格中的数据形式的相关元数据组成。表存储的数据存放在分布式文件系统里，例如HDFS，元数据存储在关系数据库里，当我们创建一张hive的表，还没有为表加载数据的时候，该表在分布式文件系统，例如hdfs上就是一个文件夹（文件目录）。Hive里的表友两种类型一种叫托管表，这种表的数据文件存储在hive的数据仓库里，一种叫外部表，这种表的数据文件可以存放在hive数据仓库外部的分布式文件系统上，也可以放到hive数据仓库里（注意：hive的数据仓库也就是hdfs上的一个目录，这个目录是hive数据文件存储的默认路径，它可以在hive的配置文件里进行配置，最终也会存放到元数据库里）。

下面是创建托管表的实例语句：

 

Create  table  tuoguan_tbl (flied string);
Load  data local  inpath ‘home/hadoop/test.txt’ into  table  tuoguan_tbl;

　　

外部表创建的实例：

Create  external table  external_tbl (flied string)
     Location  ‘/home/hadoop/external_table’;
Load  data local  inpath ‘home/hadoop/test.txt’ into  table  external_tbl;

　　

　　大家看到了创建外部表时候table之前要加关键字external，同时还要用location命令指定文件存储的路径，如果不使用locaction数据文件也会放置到hive的数据仓库里。

　　这两种表在使用的区别主drop命令上，drop是hive删除表的命令，托管表执行drop命令的时候，会删除元数据和存储的数据，而外部表执行drop命令时候只删除元数据库里的数据，而不会删除存储的数据。另外我还要谈谈表的load命令，hive加载数据时候不会对元数据进行任何检查，只是简单的移动文件的位置，如果源文件格式不正确，也只有在做查询操作时候才能发现，那个时候错误格式的字段会以NULL来显示。

 3.分区（partition）：hive里分区的概念是根据“分区列”的值对表的数据进行粗略划分的机制，在hive存储上就体现在表的主目录（hive的表实际显示就是一个文件夹）下的一个子目录，这个文件夹的名字就是我们定义的分区列的名字，没有实际操作经验的人可能会认为分区列是表的某个字段，其实不是这样，分区列不是表里的某个字段，而是独立的列，我们根据这个列存储表的里的数据文件。使用分区是为了加快数据分区的查询速度而设计的，我们在查询某个具体分区列里的数据时候没必要进行全表扫描。下面我就举一个分区使用的实例：

创建分区：

Create  table  logs(ts bigint ,line string)
Partitioned by  (dt string,country string);

　　

 

加载数据：

Local  data local  inpath ‘/home/hadoop/par/file01.txt’ into  table  logs partition (dt=’2012-06-02’,country=’cn’);

　　

 

在hive数据仓库里实际存储的路径如下所示：

/ user /hive/warehouse/logs/dt=2013-06-02/country=cn/file1.txt
/ user /hive/warehouse/logs/dt=2013-06-02/country=cn/file2.txt
/ user /hive/warehouse/logs/dt=2013-06-02/country=us/file3.txt
/ user /hive/warehouse/logs/dt=2013-06-02/country=us/file4.txt

　　

 

我们看到在表logs的目录下有了两层子目录dt=2013-06-02和country=cn

查询操作：

Select  ts,dt,line  from  logs where  country=’cn’,

　　

这个时候我们的查询操作只会扫描file1.txt和file2.txt文件。

4.桶（bucket）：上面的table和partition都是目录级别的拆分数据，bucket则是对数据源数据文件本身来拆分数据。使用桶的表会将源数据文件按一定规律拆分成多个文件，要使用bucket，我们首先要打开hive对桶的控制，命令如下：

set  hive.enforce.bucketing = true

　　

下面这段文字是我引用博客园里风生水起的博文：

示例：
建临时表student_tmp，并导入数据：
hive> desc  student_tmp;        
OK
id      int
age     int
name     string
stat_date       string
Time  taken: 0.106 seconds
hive> select  * from  student_tmp;
OK
1       20      zxm     20120801
2       21      ljz     20120801
3       19      cds     20120801
4       18      mac     20120801
5       22      android 20120801
6       23      symbian 20120801
7       25      wp      20120801
Time  taken: 0.123 seconds
 
建student表：
hive> create  table  student(id INT , age INT , name  STRING)
        >partitioned by (stat_date STRING)
        >clustered by (id) sorted by (age) into  2 bucket
        >row format delimited fields terminated by  ',' ;
 
设置环境变量：
        > set  hive.enforce.bucketing = true ;
 
插入数据：
 
        > from  student_tmp
        > insert  overwrite table  student partition(stat_date= "20120802" )
        > select  id,age, name  where  stat_date= "20120801"  sort by  age;
 
查看文件目录：
$ hadoop fs -ls / user /hive/warehouse/studentstat_date=20120802/
Found 2 items
-rw-r --r--   1 work supergroup         31 2012-07-31 19:52 /user/hive/warehouse/student/stat_date=20120802/000000_0
-rw-r --r--   1 work supergroup         39 2012-07-31 19:52 /user/hive/warehouse/student/stat_date=20120802/000001_0

　　

　　物理上，每个桶就是表（或分区）目录里的一个文件，桶文件是按指定字段值进行hash，然后除以桶的个数例如上面例子2，最后去结果余数，因为整数的hash值就是整数本身，上面例子里，字段hash后的值还是字段本身，所以2的余数只有两个0和1，所以我们看到产生文件的后缀是*0_0和*1_0,文件里存储对应计算出来的元数据。

　　Hive的桶，我个人认为没有特别的场景或者是特别的查询，我们可以没有必要使用，也就是不用开启hive的桶的配置。因为桶运用的场景有限，一个是做map连接的运算，我在后面的文章里会讲到，一个就是取样操作了，下面还是引用风生水起博文里的例子：

查看sampling数据：
hive> select  * from  student tablesample(bucket 1 out  of  2 on  id);                                                                              
Total MapReduce jobs = 1
Launching Job 1 out  of  1
.......
OK
4       18      mac     20120802
2       21      ljz     20120802
6       23      symbian 20120802
Time  taken: 20.608 seconds
 
tablesample是抽样语句，语法：TABLESAMPLE(BUCKET x OUT  OF  y)
y必须是 table 总bucket数的倍数或者因子。hive根据y的大小，决定抽样的比例。例如， table 总共分了64份，当y=32时，抽取 (64/32=)2个bucket的数据，当y=128时，抽取(64/128=)1/2个bucket的数据。x表示从哪个bucket开始抽取。例 如， table 总bucket数为32，tablesample(bucket 3 out  of  16)，表示总共抽取（32/16=）2个bucket的数据，分别为第3个bucket和第（3+16=）19个bucket的数据。

　　

　　好了，今天就写到这里了，明天要上班不能在加班写文章了。这篇博文的内容并没有写完（hive存储格式没有写），因为这个章节的知识非常重要是理解hive的关键，所以要讲的细点，明天我争取写完hive存储格式的文章，后天也就是本周二，我将为我们技术部门介绍hive的相关技术，写博文算是我的预演了。

　　最后我要讲一下自己对大数据技术的看法，我觉得大数据技术是一个跨时代的技术，是互联网技术的未来，也是云计算的未来，它的深入发展不仅仅是数据处理上，也会改变整个互联网技术的生态链，包括我们使用的技术和开发语言，很庆幸亲身经历着整个伟大时代的变革，我也要展开双臂迎接这个大时代的到来。






Hive 的数据存储 

首先，Hive 没有专门的数据存储格式，也没有为数据建立索引，用户可以非常自由的组织 Hive 中的表，只需要在创建表的时候告诉 Hive 数据中的列分隔符和行分隔符，Hive 就可以解析数据。 

其次，Hive 中所有的数据都存储在 HDFS 中，Hive 中包含以下数据模型：Table，External Table，Partition，Bucket。 

Hive 中的 Table 和数据库中的 Table 在概念上是类似的，每一个 Table 在 Hive 中都有一个相应的目录存储数据。例如，一个表 pvs，它在 HDFS 中的路径为：/wh/pvs，其中，wh 是在 hive-site.xml 中由 ${hive.metastore.warehouse.dir} 指定的数据仓库的目录，所有的 Table 数据（不包括 External Table）都保存在这个目录中。 
Partition 对应于数据库中的 Partition 列的密集索引，但是 Hive 中 Partition 的组织方式和数据库中的很不相同。在 Hive 中，表中的一个 Partition 对应于表下的一个目录，所有的 Partition 的数据都存储在对应的目录中。例如：pvs 表中包含 ds 和 city 两个 Partition，则对应于 ds = 20090801, ctry = US 的 HDFS 子目录为：/wh/pvs/ds=20090801/ctry=US；对应于 ds = 20090801, ctry = CA 的 HDFS 子目录为；/wh/pvs/ds=20090801/ctry=CA 
Buckets 对指定列计算 hash，根据 hash 值切分数据，目的是为了并行，每一个 Bucket 对应一个文件。将 user 列分散至 32 个 bucket，首先对 user 列的值计算 hash，对应 hash 值为 0 的 HDFS 目录为：/wh/pvs/ds=20090801/ctry=US/part-00000；hash 值为 20 的 HDFS 目录为：/wh/pvs/ds=20090801/ctry=US/part-00020 
External Table 指向已经在 HDFS 中存在的数据，可以创建 Partition。它和 Table 在元数据的组织上是相同的，而实际数据的存储则有较大的差异。 
Table 的创建过程和数据加载过程（这两个过程可以在同一个语句中完成），在加载数据的过程中，实际数据会被移动到数据仓库目录中；之后对数据对访问将会直接在数据仓库目录中完成。删除表时，表中的数据和元数据将会被同时删除。 
External Table 只有一个过程，加载数据和创建表同时完成（CREATE EXTERNAL TABLE ……LOCATION），实际数据是存储在 LOCATION 后面指定的 HDFS 路径中，并不会移动到数据仓库目录中。 




底层文件格式




hive有textFile,SequenceFile,RCFile三种文件格式。

其中textfile为默认格式，建表时不指定默认为这个格式，导入数据时会直接把数据文件拷贝到hdfs上不进行处理。

 

SequenceFile,RCFile格式的表不能直接从本地文件导入数据，数据要先导入到textfile格式的表中，然后再从textfile表中用insert导入到SequenceFile,RCFile表中。

写道
create table zone0000tf(ra int, dec int, mag int) row format delimited fields terminated by '|';
create table zone0000rc(ra int, dec int, mag int) row format delimited fields terminated by '|' stored as rcfile;

load data local inpath '/home/cq/usnoa/zone0000.asc ' into table zone0000tf;
insert overwrite table zone0000rc select * from zone0000tf;(begin a job)
 

 

File Format
 

TextFile

SequenceFIle

RCFFile

Data type

Text Only

Text/Binary

Text/Binary

Internal Storage Order

Row-based

Row-based

Column-based

Compression

File Based

Block Based

Block Based

Splitable

YES

YES

YES

Splitable After Compression

No

YES

YES

 
 
 

源数据放在test1表中，大小 26413896039 Byte。


创建sequencefile 压缩表test2，使用insert  overwrite table test2 select ...语句将test1数据导入 test2 ，设置配置项：

set hive.exec.compress.output=true;
set mapred.output.compress=true;
set mapred.output.compression.codec=com.hadoop.compression.lzo.LzoCodec;
SET io.seqfile.compression.type=BLOCK;
set io.compression.codecs=com.hadoop.compression.lzo.LzoCodec;

导入耗时：98.528s。另压缩类型使用默认的record，耗时为418.936s。


创建rcfile 表test3 ，同样方式导入test3。

set hive.exec.compress.output=true;
set mapred.output.compress=true;
set mapred.output.compression.codec=com.hadoop.compression.lzo.LzoCodec;
set io.compression.codecs=com.hadoop.compression.lzo.LzoCodec;

导入耗时 253.876s。


以下为其他统计数据对比：



rows类型合并耗时文件数总数据大小count(1)基于domain、referer求点击的top100
238610458原始数据11342641389603966.297s
238610458seq98.528(block) 418.936(record)11343225297382641.578394.949s（读入数据：32,253,519,280，读入行数:238610458）
238610458rcfile253.876 s15376548178129.318286.588s（读入数据：1,358,993,读入行数:238610458

因为原始数据中均是小文件，所以合并后文件数大量减少，但是hive实现的seqfile 处理竟然还是原来的数目。rcfile 使用lzo 压缩效果明显，7倍的压缩比率。查询数据中读入数据因为这里这涉及小部分数据，所以rcfile的表读入数据仅是seqfile的4%.而读入行数一致。

 
 

SequeceFile是Hadoop API提供的一种二进制文件支持。这种二进制文件直接将<key, value>对序列化到文件中。一般对小文件可以使用这种文件合并，即将文件名作为key，文件内容作为value序列化到大文件中。这种文件格式有以下好处：
1)支持压缩，且可定制为基于Record或Block压缩（Block级压缩性能较优） 
2)本地化任务支持：因为文件可以被切分，因此MapReduce任务时数据的本地化情况应该是非常好的。 
3)难度低：因为是Hadoop框架提供的API，业务逻辑侧的修改比较简单。
坏处是需要一个合并文件的过程，且合并后的文件将不方便查看。

SequenceFile 是一个由二进制序列化过的key/value的字节流组成的文本存储文件，它可以在map/reduce过程中的input/output 的format时被使用。在map/reduce过程中，map处理文件的临时输出就是使用SequenceFile处理过的。
SequenceFile分别提供了读、写、排序的操作类。
SequenceFile的操作中有三种处理方式：
1） 不压缩数据直接存储。 //enum.NONE
2） 压缩value值不压缩key值存储的存储方式。//enum.RECORD
3）key/value值都压缩的方式存储。//enum.BLOCK

 

 

工作中用到了RcFile来存储和读取RcFile格式的文件，记录下。 
RcFile是FaceBook开发的一个集行存储和列存储的优点于一身，压缩比更高，读取列更快，它在MapReduce环境中大规模数据处理中扮演着重要的角色。 
读取操作： 

Java代码  复制代码  收藏代码
  1. job信息：   
  2.         Job job = new Job();   
  3.             job.setJarByClass(类.class);   
  4.         //设定输入文件为RcFile格式   
  5.             job.setInputFormatClass(RCFileInputFormat.class);     
  6.         //普通输出   
  7.             job.setOutputFormatClass(TextOutputFormat.class);   
  8.         //设置输入路径   
  9.             RCFileInputFormat.addInputPath(job, new Path(srcpath));   
  10.             //MultipleInputs.addInputPath(job, new Path(srcpath), RCFileInputFormat.class);   
  11.         // 输出   
  12.             TextOutputFormat.setOutputPath(job, new Path(respath));   
  13.             // 输出key格式   
  14.             job.setOutputKeyClass(Text.class);     
  15.         //输出value格式   
  16.             job.setOutputValueClass(NullWritable.class);     
  17.         //设置mapper类   
  18.             job.setMapperClass(ReadTestMapper.class);   
  19.         //这里没设置reduce，reduce的操作就是读Text类型文件，因为mapper已经给转换了。   
  20.                
  21.             code = (job.waitForCompletion(true)) ? 0 : 1;   
  22.   
  23.   
  24. // mapper 类   
  25.   
  26. pulic class ReadTestMapper extends Mapper<LongWritable, BytesRefArrayWritable, Text, NullWritable> {   
  27.            
  28.         @Override  
  29.         protected void map(LongWritable key, BytesRefArrayWritable value, Context context) throws IOException, InterruptedException {   
  30.             // TODO Auto-generated method stub   
  31.             Text txt = new Text();    
  32.         //因为RcFile行存储和列存储，所以每次进来的一行数据，Value是个列簇，遍历，输出。   
  33.             StringBuffer sb = new StringBuffer();   
  34.             for (int i = 0; i < value.size(); i++) {   
  35.                 BytesRefWritable v = value.get(i);   
  36.                 txt.set(v.getData(), v.getStart(), v.getLength());   
  37.                 if(i==value.size()-1){   
  38.                     sb.append(txt.toString());   
  39.                 }else{   
  40.                     sb.append(txt.toString()+"\t");   
  41.                 }   
  42.             }   
  43.             context.write(new Text(sb.toString()),NullWritable.get());   
  44.             }   
  45.         }  
[java]  view plain copy print ?
  1. job信息：  
  2.         Job job = new Job();  
  3.             job.setJarByClass(类.class);  
  4.         //设定输入文件为RcFile格式  
  5.             job.setInputFormatClass(RCFileInputFormat.class);    
  6.         //普通输出  
  7.             job.setOutputFormatClass(TextOutputFormat.class);  
  8.         //设置输入路径  
  9.             RCFileInputFormat.addInputPath(job, new Path(srcpath));  
  10.             //MultipleInputs.addInputPath(job, new Path(srcpath), RCFileInputFormat.class);  
  11.         // 输出  
  12.             TextOutputFormat.setOutputPath(job, new Path(respath));  
  13.             // 输出key格式  
  14.             job.setOutputKeyClass(Text.class);    
  15.         //输出value格式  
  16.             job.setOutputValueClass(NullWritable.class);    
  17.         //设置mapper类  
  18.             job.setMapperClass(ReadTestMapper.class);  
  19.         //这里没设置reduce，reduce的操作就是读Text类型文件，因为mapper已经给转换了。  
  20.               
  21.             code = (job.waitForCompletion(true)) ? 0 : 1;  
  22.   
  23.   
  24. // mapper 类  
  25.   
  26. pulic class ReadTestMapper extends Mapper<LongWritable, BytesRefArrayWritable, Text, NullWritable> {  
  27.           
  28.         @Override  
  29.         protected void map(LongWritable key, BytesRefArrayWritable value, Context context) throws IOException, InterruptedException {  
  30.             // TODO Auto-generated method stub  
  31.             Text txt = new Text();   
  32.         //因为RcFile行存储和列存储，所以每次进来的一行数据，Value是个列簇，遍历，输出。  
  33.             StringBuffer sb = new StringBuffer();  
  34.             for (int i = 0; i < value.size(); i++) {  
  35.                 BytesRefWritable v = value.get(i);  
  36.                 txt.set(v.getData(), v.getStart(), v.getLength());  
  37.                 if(i==value.size()-1){  
  38.                     sb.append(txt.toString());  
  39.                 }else{  
  40.                     sb.append(txt.toString()+"\t");  
  41.                 }  
  42.             }  
  43.             context.write(new Text(sb.toString()),NullWritable.get());  
  44.             }  
  45.         }  



输出压缩为RcFile格式： 

Java代码  复制代码  收藏代码
  1. job信息：   
  2.             Job job = new Job();   
  3.             Configuration conf = job.getConfiguration();   
  4.         //设置每行的列簇数   
  5.             RCFileOutputFormat.setColumnNumber(conf, 4);   
  6.             job.setJarByClass(类.class);   
  7.   
  8.             FileInputFormat.setInputPaths(job, new Path(srcpath));   
  9.             RCFileOutputFormat.setOutputPath(job, new Path(respath));   
  10.   
  11.             job.setInputFormatClass(TextInputFormat.class);   
  12.             job.setOutputFormatClass(RCFileOutputFormat.class);   
  13.   
  14.             job.setMapOutputKeyClass(LongWritable.class);   
  15.             job.setMapOutputValueClass(BytesRefArrayWritable.class);   
  16.   
  17.             job.setMapperClass(OutPutTestMapper.class);   
  18.   
  19.             conf.set("date", line.getOptionValue(DATE));   
  20.         //设置压缩参数   
  21.             conf.setBoolean("mapred.output.compress", true);   
  22.             conf.set("mapred.output.compression.codec", "org.apache.hadoop.io.compress.GzipCodec");   
  23.   
  24.             code = (job.waitForCompletion(true)) ? 0 : 1;   
  25.   
  26. mapper类：   
  27.     public class OutPutTestMapper extends Mapper<LongWritable, Text, LongWritable, BytesRefArrayWritable> {   
  28.         @Override  
  29.         public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {   
  30.             String line = value.toString();   
  31.             String day = context.getConfiguration().get("date");   
  32.             if (!line.equals("")) {   
  33.                 String[] lines = line.split(" ", -1);   
  34.                 if (lines.length > 3) {   
  35.                     String time_temp = lines[1];   
  36.                     String times = timeStampDate(time_temp);   
  37.                     String d = times.substring(0, 10);   
  38.                     if (day.equals(d)) {   
  39.                         byte[][] record = {lines[0].getBytes("UTF-8"), lines[1].getBytes("UTF-8"),lines[2].getBytes("UTF-8"), lines[3].getBytes("UTF-8")};   
  40.   
  41.                         BytesRefArrayWritable bytes = new BytesRefArrayWritable(record.length);   
  42.   
  43.                         for (int i = 0; i < record.length; i++) {   
  44.                             BytesRefWritable cu = new BytesRefWritable(record[i], 0, record[i].length);   
  45.                             bytes.set(i, cu);   
  46.                         }   
  47.                         context.write(key, bytes);   
  48.                     }   
  49.                 }   
  50.             }   
  51.         }  

 


SequenceFile提供了若干Writer的构造静态获取。
//SequenceFile.createWriter()；

SequenceFile.Reader使用了桥接模式，可以读取SequenceFile.Writer中的任何方式的压缩数据。

三种不同的压缩方式是共用一个数据头，流方式的读取会先读取头字节去判断是哪种方式的压缩，然后根据压缩方式去解压缩并反序列化字节流数据，得到可识别的数据。

流的存储头字节格式：
Header：
*字节头”SEQ”, 后跟一个字节表示版本”SEQ4”,”SEQ6”.//这里有点忘了 不记得是怎么处理的了，回头补上做详细解释
*keyClass name
*valueClass name
*compression boolean型的存储标示压缩值是否转变为keys/values值了
*blockcompression boolean型的存储标示是否全压缩的方式转变为keys/values值了
*compressor 压缩处理的类型，比如我用Gzip压缩的Hadoop提供的是GzipCodec什么的..
*元数据 这个大家可看可不看的

所有的String类型的写操作被封装为Hadoop的IO API，Text类型writeString()搞定。

未压缩的和只压缩values值的方式的字节流头部是类似的：
*Header
*RecordLength记录长度
*key Length key值长度
*key 值
*是否压缩标志 boolean
*values
剩下的大家可看可不看的，并非这个类中主要的。












///文件存储格式图






整理了一下网上的几种Hive文件存储格式的性能与Hadoop的文件存储格式。

 

Hive的三种文件格式：TEXTFILE、SEQUENCEFILE、RCFILE中，TEXTFILE和SEQUENCEFILE的存储格式都是基于行存储的，RCFILE是基于行列混合的思想，先按行把数据划分成N个row group，在row group中对每个列分别进行存储。另：Hive能支持自定义格式，详情见：Hive文件存储格式

基于HDFS的行存储具备快速数据加载和动态负载的高适应能力，因为行存储保证了相同记录的所有域都在同一个集群节点。但是它不太满足快速的查询响应时间的要求，因为当查询仅仅针对所有列中的 少数几列时，它就不能跳过不需要的列，直接定位到所需列；同时在存储空间利用上，它也存在一些瓶颈，由于数据表中包含不同类型，不同数据值的列，行存储不 易获得一个较高的压缩比。RCFILE是基于SEQUENCEFILE实现的列存储格式。除了满足快速数据加载和动态负载高适应的需求外，也解决了SEQUENCEFILE的一些瓶颈。

 

下面对这几种几个作一个简单的介绍：

 

TextFile：

 

Hive默认格式，数据不做压缩，磁盘开销大，数据解析开销大。
可结合Gzip、Bzip2、Snappy等使用（系统自动检查，执行查询时自动解压），但使用这种方式，hive不会对数据进行切分，从而无法对数据进行并行操作。

 

SequenceFile：

SequenceFile是Hadoop API 提供的一种二进制文件，它将数据以<key,value>的形式序列化到文件中。这种二进制文件内部使用Hadoop 的标准的Writable 接口实现序列化和反序列化。它与Hadoop API中的MapFile 是互相兼容的。Hive 中的SequenceFile 继承自Hadoop API 的SequenceFile，不过它的key为空，使用value 存放实际的值， 这样是为了避免MR 在运行map 阶段的排序过程。

 

SequenceFile的文件结构图：

 

Header通用头文件格式：

SEQ3BYTE
Nun1byte数字
keyClassName 
ValueClassName 
compression（boolean）指明了在文件中是否启用压缩
blockCompression（boolean，指明是否是block压缩）
compressioncodec
Metadata文件元数据
Sync头文件结束标志

Block-Compressed SequenceFile格式



 

RCFile

RCFile是Hive推出的一种专门面向列的数据格式。 它遵循“先按列划分，再垂直划分”的设计理念。当查询过程中，针对它并不关心的列时，它会在IO上跳过这些列。需要说明的是，RCFile在map阶段从 远端拷贝仍然是拷贝整个数据块，并且拷贝到本地目录后RCFile并不是真正直接跳过不需要的列，并跳到需要读取的列， 而是通过扫描每一个row group的头部定义来实现的，但是在整个HDFS Block 级别的头部并没有定义每个列从哪个row group起始到哪个row group结束。所以在读取所有列的情况下，RCFile的性能反而没有SequenceFile高。

 

下面介绍行存储、列存储（详细参照：Facebook数据仓库揭秘：RCFile高效存储结构）

 

行存储

 

HDFS块内行存储的例子：

 基于Hadoop系统行存储结构的优点在于快速数据加载和动态负载的高适应能力，这是因为行存储保证了相同记录的所有域都在同一个集群节点，即同一个 HDFS块。不过，行存储的缺点也是显而易见的，例如它不能支持快速查询处理，因为当查询仅仅针对多列表中的少数几列时，它不能跳过不必要的列读取；此 外，由于混合着不同数据值的列，行存储不易获得一个极高的压缩比，即空间利用率不易大幅提高。

 

列存储

HDFS块内列存储的例子

 在HDFS上按照列组存储表格的例子。在这个例子中，列A和列B存储在同一列组，而列C和列D分别存储在单独的列组。查询时列存储能够避免读不必要的列， 并且压缩一个列中的相似数据能够达到较高的压缩比。然而，由于元组重构的较高开销，它并不能提供基于Hadoop系统的快速查询处理。列存储不能保证同一 记录的所有域都存储在同一集群节点，行存储的例子中，记录的4个域存储在位于不同节点的3个HDFS块中。因此，记录的重构将导致通过集群节点网络的大 量数据传输。尽管预先分组后，多个列在一起能够减少开销，但是对于高度动态的负载模式，它并不具备很好的适应性。

 

RCFile结合行存储查询的快速和列存储节省空间的特点：首先，RCFile保证同一行的数据位于同一节点，因此元组重构的开销很低；其次，像列存储一样，RCFile能够利用列维度的数据压缩，并且能跳过不必要的列读取。

HDFS块内RCFile方式存储的例子：



 
 数据测试

源表数据记录数： 67236221

 

第一步：创建三种文件类型的表，建表语法参考Hive文件存储格式

 

Sql代码   收藏代码
  1. --TextFile  
  2. set hive.exec.compress.output=true;  
  3. set mapred.output.compress=true;  
  4. set mapred.output.compression.codec=org.apache.hadoop.io.compress.GzipCodec;  
  5. set io.compression.codecs=org.apache.hadoop.io.compress.GzipCodec;  
  6. INSERT OVERWRITE table hzr_test_text_table PARTITION(product='xxx',dt='2013-04-22')  
  7. SELECT xxx,xxx.... FROM xxxtable WHERE product='xxx' AND dt='2013-04-22';  
  8.   
  9. --SquenceFile  
  10. set hive.exec.compress.output=true;  
  11. set mapred.output.compress=true;  
  12. set mapred.output.compression.codec=org.apache.hadoop.io.compress.GzipCodec;  
  13. set io.compression.codecs=org.apache.hadoop.io.compress.GzipCodec;  
  14. set io.seqfile.compression.type=BLOCK;  
  15. INSERT OVERWRITE table hzr_test_sequence_table PARTITION(product='xxx',dt='2013-04-22')  
  16. SELECT xxx,xxx.... FROM xxxtable WHERE product='xxx' AND dt='2013-04-22';  
  17.   
  18. --RCFile  
  19. set hive.exec.compress.output=true;  
  20. set mapred.output.compress=true;  
  21. set mapred.output.compression.codec=org.apache.hadoop.io.compress.GzipCodec;  
  22. set io.compression.codecs=org.apache.hadoop.io.compress.GzipCodec;  
  23. INSERT OVERWRITE table hzr_test_rcfile_table PARTITION(product='xxx',dt='2013-04-22')  
  24. SELECT xxx,xxx.... FROM xxxtable WHERE product='xxx' AND dt='2013-04-22';  
 

 

第二步：测试insert overwrite table tablename select.... 耗时，存储空间

类型insert耗时(S)存储空间(G)

Sequence

97.291

7.13G

RCFile

120.901

5.73G

TextFile

290.517 

6.80G

 

insert耗时、count(1)耗时比较：

 


 

第三步：查询响应时间

 

    测试一

 

Sql代码   收藏代码
  1. 方案一，测试整行记录的查询效率：  
  2. select * from hzr_test_sequence_table where game='XXX' ;    
  3. select * from hzr_test_rcfile_table where game='XXX' ;   
  4. select * from hzr_test_text_table where game='XXX' ;   
  5.   
  6. 方案二，测试特定列的查询效率：  
  7. select game,game_server from hzr_test_sequence_table where game ='XXX';  
  8. select game,game_server from hzr_test_rcfile_table where game ='XXX';  
  9. select game,game_server from hzr_test_text_table where game ='XXX';  

 

文件格式

查询整行记录耗时（S）

查询特定列记录耗时（S）

sequence

42.241

39.918

rcfile

37.395

36.248

text

43.164

41.632

 

方案耗时对比：



 

 

   测试二：

本测试目的是验证RCFILE的数据读取方式和Lazy解压方式是否有性能优势。数据读取方式只读取元数据和相关的列，节省IO；Lazy解压方式只解压相关的列数据，对不满足where条件的查询数据不进行解压，IO和效率都有优势。

 

方案一：

记录数：698020

 

Sql代码   收藏代码
  1. insert overwrite local directory 'XXX/XXXX' select game,game_server from hzr_test_xxx_table where game ='XXX';  

 

 

方案二：

记录数：67236221

 

Sql代码   收藏代码
  1. insert overwrite local directory 'xxx/xxxx' select game,game_server from hzr_test_xxx_table;  

 

方案三：

记录数：

 

Sql代码   收藏代码
  1. insert overwrite local directory 'xxx/xxx'  
  2. select game from hzr_xxx_rcfile_table;  

 

文件类型方案一方案二方案三
TextFile54.89569.428 167.667
SequenceFile137.09677.03  123.667
RCFile44.2857.037 89.9

 

 上图表现反应在大小数据集上，RCFILE的查询效率高于SEQUENCEFILE，在特定字段数据读取时，RCFILE的查询效率依然优于SEQUENCEFILE。


声明：本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：【wpsshop博客】
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号