hive存储处理器（StorageHandlers）以及hive与hbase整合

 此篇文章基于hive官方英文文档翻译，有些不好理解的地方加入了我个人的理解，官方的英文地址为：

1.https://cwiki.apache.org/confluence/display/Hive/StorageHandlers

2.https://cwiki.apache.org/confluence/display/Hive/HBaseIntegration

一 存储处理器介绍

通过HIVE存储处理器，不但可以让hive基于hbase实现，还可以支持cassandra JDBC MongoDB 以及 Google Spreadsheets 

HIVE存储器的实现原理基于HIVE以及Hadoop的可扩展性实现：

输入格式化（input formats）

输出格式化（output formats）

序列化/反序列化包（serialization/deserialization librarises） 

除了依据以上可扩展性，存储处理器还需要实现新的元数据钩子接口，这个接口允许使用HIVE的DDL语句来定义和管理hive自己的元数据以及其它系统的目录（此目录个人理解为其它系统的元数据目录）

一些术语：

HIVE本身有的概念：

被管理的表即内部表（managed）：元数据由hive管理，并且数据也存储在hive的体系里面

外部表（external table）：表的定义被外部的元数据目录所管理，数据也存储在外部系统中

hive存储处理器的概念：

本地（native）表：hive不需要借助存储处理器就可以直接管理和访问的表

非本地（non-native）表:需要通过存储处理器才能管理和访问的表

内部表 外部表 和 本地表 非本地表 形成交叉，就有了下面四种形式的概念定义：

被管理的本地表（managed native）：通过CREATE TABLE创建的表

外部本地表(external native)：通过CREATE EXTERNAL TABLE创建，但是没有带STORED BY子句

被管理的非本地表（）managed non-native):通过CREATE TABLE 创建，同时有STORED BY子句，hive在元数据中存储表定义，但是不创建任何文件（我的理解是存放数据的目录文件，hive在定义好表结构后会创建对应的目录来存储对应的数据），hive存储处理器向存储数据的系统发出一个请求来创建一个一致的对象结构；

外部非本地（external non-native）：通过CREATE EXTERNAL TABLE创建，并且带有STORED BY子句；hive在自己的元数据中注册表的定义信息，并且通过调用存储处理器来检查这些注册在hive中的信息是否与其它系统中原来定义的信息一致

通过hive创建表结构的DDL语句：

CREATE [EXTERNAL] TABLE [IF NOT EXISTS] table_name

  [(col_name data_type [COMMENT col_comment], ...)]

  [COMMENT table_comment]

  [PARTITIONED BY (col_name data_type [col_comment], col_name data_type [COMMENT col_comment], ...)]

  [CLUSTERED BY (col_name, col_name, ...) [SORTED BY (col_name, ...)] INTO num_buckets BUCKETS]

  [

   [ROW FORMAT row_format] [STORED AS file_format]

   | STORED BY 'storage.handler.class.name' [WITH SERDEPROPERTIES (...)]

  ]

  [LOCATION hdfs_path]

  [AS select_statement]

具体的使用样例，会在下面详细说明

二 Hive与Hbase的整合

hive和Hbase的整合通过Hbase handler jar包实现，它的形式为hive-hbase-x.y.z.jar ,这个处理器需要依赖hadoop0.20以上版本，并且只在hadoop-0.20.x hbase-0.92.0 和zookeeper-3.3.4 上进行过测试。如果hbase版本不是0.92则需要基于你使用的版本重新编译对应的hive存储处理器。

为了创建一张可以被hive管理的hbase的表，需要在hive的ddl语句CREATE TABLE 后面加入语句STORED BY

CREATE TABLE hbase_table_1(key int, value string)

STORED BY 'org.apache.hadoop.hive.hbase.HBaseStorageHandler'

WITH SERDEPROPERTIES ("hbase.columns.mapping" = ":key,cf1:val")

TBLPROPERTIES ("hbase.table.name" = "xyz");

Hbase.table.name属性是可选的，用它来指定此表在hbase中的名字，这就是说，允许同一个表在hive和hbase中有不同的名字。上述语句中，在hive中的表名叫hbase_talbe_1,在hbase中，此表名叫xyz。如果不指定，两个名字则会是相同的。

当执行完命令后，就可以在HBase的shell中看到一张新的空表，如下：

$ hbase shell

HBase Shell; enter 'help<RETURN>' for list of supported commands.

Version: 0.20.3, r902334, Mon Jan 25 13:13:08 PST 2010

hbase(main):001:0> list

xyz                                                                                                          

1 row(s) in 0.0530 seconds

hbase(main):002:0> describe "xyz"

DESCRIPTION                                                             ENABLED                              

 {NAME => 'xyz', FAMILIES => [{NAME => 'cf1', COMPRESSION => 'NONE', VE true                                 

 RSIONS => '3', TTL => '2147483647', BLOCKSIZE => '65536', IN_MEMORY =>                                      

  'false', BLOCKCACHE => 'true'}]}                                                                           

1 row(s) in 0.0220 seconds

hbase(main):003:0> scan "xyz"

ROW                          COLUMN+CELL                                                                     

0 row(s) in 0.0060 seconds

上面的输出中没有列val的信息，虽然在建表语句中已经指定了这个列名。这是因为在hbase中只有列族的名字才会被定义在表级的元数据中，列族中的列只定义在行级别的元数据中。

下面的语句是定义怎么把数据从hive中加载进入HBase的表，表pokes是hive中已经存在的一个表，并且表中有数据。

INSERT OVERWRITE TABLE hbase_table_1 SELECT * FROM pokes WHERE foo=98;

然后，在Hbase shell中验证数据是否已经加载：

hbase(main):009:0> scan "xyz"

ROW                          COLUMN+CELL                                                                     

 98                          column=cf1:val, timestamp=1267737987733, value=val_98                           

1 row(s) in 0.0110 seconds

通过hive的语句查询表的结果如下：

hive> select * from hbase_table_1;

Total MapReduce jobs = 1

Launching Job 1 out of 1

...

OK

98        val_98

Time taken: 4.582 seconds

Inserting large amounts of data may be slow due to WAL overhead; if you would like to disable this, make sure you have HIVE-1383 (as of Hive 0.6), and then issue this command before the INSERT:

set hive.hbase.wal.enabled=false;

Warning: disabling WAL may lead to data loss if an HBase failure occurs, so only use this if you have some other recovery strategy available.

如果想基于已经存在的hbase表创建hive可访问的表，则需要用CREATE EXTERNAL TABLE，如下：

CREATE EXTERNAL TABLE hbase_table_2(key int, value string)

STORED BY 'org.apache.hadoop.hive.hbase.HBaseStorageHandler'

WITH SERDEPROPERTIES ("hbase.columns.mapping" = "cf1:val")

TBLPROPERTIES("hbase.table.name" = "some_existing_table");

hbase.columns.mapping属性项是必须存在的，它代表的是hbase表的列族以及列的定义信息。

hive与hbase的列匹配

有两种SERDEPROPERTIES配置信息来控制HBase列到HIve的映射

1.hbase.columns,mapping

2.hbase.table.default.storage.type

  这个属性可以有String和二进制两种类型的值，默认是String；这个选项只有在hive0.9这个版本可用，     并且在早期的版本中，只有String类型可用

当前列匹配使用起来显得有点笨重和憋手：

1.每个hive的列，都需要在参数hbase.columns.mapping中指定一个对应的条目（比如：a：b或者：key就叫一个条目），多个列之间的条目通过逗号分隔；也就是说，如果某个表有n个列，则参数hbase.columns.mapping的值中就有n个以逗号分隔的条目。比如：

"hbase.columns.mapping" = ":key,a:b,a:c,d:e" 代表有两个列族，一个是a一个是d，a列族中有两列，分别为b和c

注意，hbase.columns.mapping的值中是不允许出现空格的

2.每个匹配条目的形式必须是：key或者列族名：[列名][#(binary|string)](前面带有#标识的是在hive0.9才添加的，早期的版本把所有都当作String类型)

       1）如果没有给列或者列族指定类型，hbase.table.default.storage.type的值会被当作这些列或者列族的类  型

        2）表示类型的值（binary|string）的任何前缀都可以被用来表示这个类型，比如#b就代表了#binary

        3）如果指定列的类型为二进制（binary）字节，则在HBase的单元格的存储类型也必须为二进制字节

3.必须窜至一个：key的形式匹配条目（不支持复合key的形式）

4.在hive0.6版本以前，是通过第一个条目来作为关键字（key）字段，从0.6以后，都需要直接通过：key的方式来指定

5.当前没有办法可以访问到HBase的时间撮属性，并且查询总是访问到最新的数据（这条主要是因为hbase单元格数据存储是有版本的，根据时间撮）

6. 因为HBase的列定义没有包含有数据类型信息，所以在存储的时候，会把所有的其它类型都转换为string 代表；所以，列数据类型不支持自定义

7. 没有必要对HBase的所有列族都进行映射，但是没有被映射的列族不能通过访问Hive表读取到数据；可以把多个Hive表映射到同一个HBase的表

下面的章节提供更加详细的例子，来说明当前各种不同的列映射

多个列和列族

下面的例子包含三个Hive表的列以及两个HBase表的列族，其中一个列族包括两个列

CREATE TABLE hbase_table_1(key int, value1 string, value2 int, value3 int)

STORED BY 'org.apache.hadoop.hive.hbase.HBaseStorageHandler'

WITH SERDEPROPERTIES (

"hbase.columns.mapping" = ":key,a:b,a:c,d:e"

);

INSERT OVERWRITE TABLE hbase_table_1 SELECT foo, bar, foo+1, foo+2

FROM pokes WHERE foo=98 OR foo=100;

上面的例子中，hive拥有除去可以外的列为三个，value1 value2 value3，对应的HBase表为两个列族，分别为a和d，其中a包括两列（b和c）；值的对应关系从左到右一一对应，a列族中的b列对应value1，c列对应value2，d列族的e列对应value3 

下面是从HBase中查看的结果：

hbase(main):014:0> describe "hbase_table_1"

DESCRIPTION                                                             ENABLED                              

 {NAME => 'hbase_table_1', FAMILIES => [{NAME => 'a', COMPRESSION => 'N true                                 

 ONE', VERSIONS => '3', TTL => '2147483647', BLOCKSIZE => '65536', IN_M                                      

 EMORY => 'false', BLOCKCACHE => 'true'}, {NAME => 'd', COMPRESSION =>                                       

 'NONE', VERSIONS => '3', TTL => '2147483647', BLOCKSIZE => '65536', IN                                      

 _MEMORY => 'false', BLOCKCACHE => 'true'}]}                                                                 

1 row(s) in 0.0170 seconds

hbase(main):015:0> scan "hbase_table_1"

ROW                          COLUMN+CELL                                                                     

 100                         column=a:b, timestamp=1267740457648, value=val_100                              

 100                         column=a:c, timestamp=1267740457648, value=101                                  

 100                         column=d:e, timestamp=1267740457648, value=102                                  

 98                          column=a:b, timestamp=1267740457648, value=val_98                               

 98                          column=a:c, timestamp=1267740457648, value=99                                   

 98                          column=d:e, timestamp=1267740457648, value=100                                  

2 row(s) in 0.0240 seconds

同一张表在Hive中的查询结果是：

hive> select * from hbase_table_1;

Total MapReduce jobs = 1

Launching Job 1 out of 1

...

OK

100        val_100        101        102

98        val_98        99        100

Time taken: 4.054 seconds

Hive MAP(集合)与Hbase 列族的映射

下面是Hive的MAP数据类型与Hbase列族的映射例子。每行都可以包含不同的列组合，列名与map的可以对应，values与列值对应。

CREATE TABLE hbase_table_1(value map<string,int>, row_key int)

STORED BY 'org.apache.hadoop.hive.hbase.HBaseStorageHandler'

WITH SERDEPROPERTIES (

"hbase.columns.mapping" = "cf:,:key"

);

INSERT OVERWRITE TABLE hbase_table_1 SELECT map(bar, foo), foo FROM pokes

WHERE foo=98 OR foo=100;

其中cf表示列族，冒号后面为空，它与Hive表的列value对应，即value的key为cf列族的列，可以通过下面HBase的查询来理解key为列名，value为值的用法。

hbase(main):012:0> scan "hbase_table_1"

ROW                          COLUMN+CELL                                                                     

 100                         column=cf:val_100, timestamp=1267739509194, value=100                           

 98                          column=cf:val_98, timestamp=1267739509194, value=98                             

2 row(s) in 0.0080 seconds

cf为列族名，val_100为hive表中MAP（集合）的key，100为MAP（集合）中的val_100的值

下面是对应的hive的查询显示结果：

hive> select * from hbase_table_1;

Total MapReduce jobs = 1

Launching Job 1 out of 1

...

OK

{"val_100":100}        100

{"val_98":98}        98

Time taken: 3.808 seconds

注意：MAP（集合）的key必须是string类型，否则会失败，因为key是HBase列的名字；如下的定义将会失败

CREATE TABLE hbase_table_1(key int, value map<int,int>)

STORED BY 'org.apache.hadoop.hive.hbase.HBaseStorageHandler'

WITH SERDEPROPERTIES (

"hbase.columns.mapping" = ":key,cf:"

);

FAILED: Error in metadata: java.lang.RuntimeException: MetaException(message:org.apache.hadoop.hive.serde2.SerDeException org.apache.hadoop.hive.hbase.HBaseSerDe: hbase column family 'cf:' should be mapped to map<string,?> but is mapped to map<int,int>)

注意：当hbase.columns.mapping中有“：key,cf:”这样的值，即列族冒号后面为空时，表示Hive中对应的类型为集合map，如果不是，则创建表会失败

CREATE TABLE hbase_table_1(key int, value string)

STORED BY 'org.apache.hadoop.hive.hbase.HBaseStorageHandler'

WITH SERDEPROPERTIES (

"hbase.columns.mapping" = ":key,cf:"

);

FAILED: Error in metadata: java.lang.RuntimeException: MetaException(message:org.apache.hadoop.hive.serde2.SerDeException

org.apache.hadoop.hive.hbase.HBaseSerDe: hbase column family 'cf:' should be mapped to map<string,?> but is mapped to string)

hbase.columns.mapping中值类型用法列举

如果没有指定值的类型，比如cf：val，则采用配置hbase.table.default.storage.type的值为数据类型

1.当不配置hbase.table.default.storage.type时，它默认是string，如果有二进制的数据类型，则如下定义：

CREATE TABLE hbase_table_1 (key int, value string, foobar double)

STORED BY 'org.apache.hadoop.hive.hbase.HBaseStorageHandler'

WITH SERDEPROPERTIES (

"hbase.columns.mapping" = ":key#b,cf:val,cf:foo#b"

);

2.如果显示的指定hbase.table.default.storage.type为binary时，如果类型有string类型，则需要指定，如：

CREATE TABLE hbase_table_1 (key int, value string, foobar double)

STORED BY 'org.apache.hadoop.hive.hbase.HBaseStorageHandler'

WITH SERDEPROPERTIES (

"hbase.columns.mapping" = ":key,cf:val#s,cf:foo",

"hbase.table.default.storage.type" = "binary"

);

 cf:val#s 中的#s就表示类型为string，cf:foo没有配置类型，则采用hbase.table.default.storage.type的配置，为binary

添加时间戳

当用hive给HBase的表添加记录时，时间戳默认为当前时间，如果想改变这个值，可以通过设置SERDEPROPERIES属性的可选配置项hbase.put.timestamp,当这个配置项为-1的时候，就是默认策略，即添加记录为当前时间戳

CREATE TABLE hbase_table_1 (key int, value string, foobar double)

STORED BY 'org.apache.hadoop.hive.hbase.HBaseStorageHandler'

WITH SERDEPROPERTIES (

"hbase.columns.mapping" = ":key,cf:val#s,cf:foo",

"hbase.table.default.storage.type" = "binary"

"hbase.put.timestamp" = "2013-03-17 09:04:30"

)

主键唯一性

HBase与Hive的表有一个细微的差别，就是HBase的表有一个主键，并且需要唯一，但是Hive没有；如果不能保证这个主键的唯一，则HBase存储的时候，只能存储其中一个，这会导致查询的时候，hive总能找到正确的值，而HBase出来的结果就不确定

如下，在hive中查询：

CREATE TABLE pokes2(foo INT, bar STRING);

INSERT OVERWRITE TABLE pokes2 SELECT * FROM pokes;

-- this will return 3

SELECT COUNT(1) FROM POKES WHERE foo=498;

-- this will also return 3

SELECT COUNT(1) FROM pokes2 WHERE foo=498;

同一张表在HBase中查询，就会出来不确定的结果：

CREATE TABLE pokes3(foo INT, bar STRING)

STORED BY 'org.apache.hadoop.hive.hbase.HBaseStorageHandler'

WITH SERDEPROPERTIES (

"hbase.columns.mapping" = ":key,cf:bar"

);

INSERT OVERWRITE TABLE pokes3 SELECT * FROM pokes;

-- this will return 1 instead of 3

SELECT COUNT(1) FROM pokes3 WHERE foo=498;

覆盖（Overwrite）

当执行OVERWRITE时，HBase中已经存在的记录是不会被删除的；但是，如果存在的记录与新纪录的主键（key）是一样的，则老数据会被覆盖