HBase与Hive入门研究

时间：2022-02-13 08:25:42

Hadoop学习文档

Hbase学习总结

  1. Hbase的安装与运行

 

1.下载并解压hbase-0.94.6.1.tar.gz

 

tar xfz hbase-0.94.6.1.tar.gz

2.编辑conf/hbase-site.xml

在configuration标签中添加：

<property>

<name>hbase.rootdir</name>

<value>file:///var/lib/tcommsvr/Hbase_lin/hbase</value>

</property>

说明：该项配置了数据写入的目录，默认 hbase.rootdir 是指向 /tmp/hbase-${user.name} ，也就说你会在重启后丢失数据(重启的时候操作系统会清理/tmp目录)。

3.编辑conf/hbase-env.sh

修改JAVA_HOME路径

 

export JAVA_HOME=/usr/java/jdk1.6.0_29

说明：该项默认是注释掉的，需要放开并修改

 

4.进入bin目录，运行start-hbase.sh

进入logs目录下查看hbase-root-master-slave2.log日志，查看运行情况

5.查看hbase文件夹，可以看到

HBase与Hive入门研究

如上图所示的内容生成。

 

Hbase的shell操作

1.进入bin目录，运行sh hbase shell进入shell命令

输入help可查看常用命令

下面测试建立一个表并添加几条数据，并查询，然后删除该表

 

create 'test_lin' ,'cf'

 

然后执行list,可以看到该表已创建，该表名称为test_lin , 这个表只有一个column family 为 cf,注意：表名称和column family 都必须加上单引号

 

下面插入一些记录：

 

put 'test_lin','lucy','cf:name_lucy','New York_lucy'

put 'test_lin','lily','cf:name_lily','London_Lily'

put 'test_lin','Jack','cf:name_Jack','Beijing_Jack'

 

查看该表：scan 'test_lin'

 

获取一行（行为Jack的）数据

 

get 'test_lin', 'Jack'

 

清除该表（删掉）

 

disable 'test_lin'

然后

drop 'test_lin'

 

关闭shell：exit

 

停止Hbase

 

Bin目录中有相应的脚本：stop-hbase.sh

 

  1. Hbase的存储结构

 

数据在Hbase中是以表的方式存储的，所有的表（由行和列构成）从属于某一个column family。行和列的交叉点称为cell，cell是版本化的。Cell的内容是不可分割的字节数组

表的行的key也是一段字节数组。Hbase的表是按照key排序的，排序方式是对字节的。所有的表都必须要有主键key

以上文为例：该表test_lin中有一个column family ,但是该column family中却有三列，name_lucy,name_lily,name_jack，如果应用于我们的系统，此处应该是一列

 

 

该表的实际存储结构图如下（可执行scan 'test_lin'看到）：

ROW COLUMN+CELL

Jack column=cf:name_Jack, timestamp=1365514700679, value=Beijing_Jack

lily column=cf:name_lily, timestamp=1365514671563, value=London_Lily

lucy column=cf:name_lucy, timestamp=1365506747279, value=New York_lucy

 

可以注意到，该表在创建的时候并未声明column的名字，也就是说当心的column产生的时候，可以不经过column声明而直接插入到column family中。但是column family 则必须在创建表结构的时候定义

 

关于HBase与Hive入门研究

此表中，如果我们根据row key为 'com.cnn.www'获取数据的话我们会得到时间戳为t9中anchor:cnnsi.com="CNN" ，t8中anchor:my.look.ca="CNN.com"，t6中contents:html="<html>…"的数据。这是因为如果不指明行，默认获取最新时间的数据

关于column family的几点说明：

1.一个column family所有列成员有相同的前缀，如上文创建的表，前缀都是cf,列名是冒号后面的数据。

2.一个column family所有成员在文件系统中都是存储在一起的，存储优化是针对column family级别的，也就意味着一个column family的所有成员都是用相同的方式访问的。

 

A {row, column, version} 元组就是一个Hbase中的一个 cell。Cell的内容是不可分割的字节数组。这一点类似于我们当前根据车辆以及时间进行分表处理。（此处可把一个元组看成是一辆车的一条记录，该记录是不可分割的。）

  1. 本地Hbase实例

 

本地连接Hbase做开发：

 

说明：此为单独测试Hbase数据库的DEMO，不做集群。

 

（1）Hbase地址：119.255.194.51 路径：/var/lib/tcommsvr/Hadoop_lin/hbase-0.94.6.1

（2）conf 中修改

hbase-site.xml

 

<property>

        <name>hbase.rootdir</name>

        <value>file:///var/lib/tcommsvr/Hadoop_lin/hbase</value>

    </property>

    <property>

        <name>hbase.zookeeper.quorum</name>

        <value>10.10.10.21</value>

    </property>

hbase-env.sh修改

export JAVA_HOME=/usr/local/jdk1.6.0_17

 

（3）启动Hhbase

（4）配置本地网络环境，使本机与Hbase处于相同局域网

（5）本地工程结构：

HBase与Hive入门研究

 

/*

* 版权信息：北京赛德斯汽车信息技术有限公司</br>

* Copyright ?2010-2012. All rights reserved. 京ICP备11043401号

*/

package com.sides.hbase.demo;

 

import java.io.IOException;

import java.util.ArrayList;

import java.util.List;

 

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.hbase.HBaseConfiguration;

import org.apache.hadoop.hbase.HColumnDescriptor;

import org.apache.hadoop.hbase.HTableDescriptor;

import org.apache.hadoop.hbase.KeyValue;

import org.apache.hadoop.hbase.client.Delete;

import org.apache.hadoop.hbase.client.Get;

import org.apache.hadoop.hbase.client.HBaseAdmin;

import org.apache.hadoop.hbase.client.HTable;

import org.apache.hadoop.hbase.client.Put;

import org.apache.hadoop.hbase.client.Result;

import org.apache.hadoop.hbase.client.ResultScanner;

import org.apache.hadoop.hbase.client.Scan;

import org.apache.hadoop.hbase.util.Bytes;

 

 

 

/****************************************************************************

* com.sides.hbase.demo HBaseDemo.java Created on 2013-6-27

*

* @Author: linfenliang

* @Description:HBase设计模式：

* <pre>

* 1.列族的数量越少越好，考虑同一个表中不同列族所存储的记录数量的差别(列族的势)[查询或scan效率会有一定影响];

* 2.避免使用时序或单调(递增或递减)的行键(Row Key);

* 3.尽量最小化行键和列族的大小[方便索引];

* 4.HBase默认保存三个版本，不要设置的过大

* </pre>

* @Version: 1.0

***************************************************************************/

public class HBaseDemo {

    private static Configuration conf = null;

    static{

        Configuration hbaseConfig = new Configuration();

        hbaseConfig.set("hbase.zookeeper.quorum", "10.10.10.21");

//        hbaseConfig.set("hbase.zookeeper.property.clientPort", "2181");

        conf = HBaseConfiguration.create(hbaseConfig);    //手动配置初始化参数

//        HBaseConfiguration.create()//该方法将自动从classpath中查找hbase-site.xml初始化Configuration

    }

    /**

     * 创建表

     * @param tableName

     * @param familys

     * @throws IOException

     * void

     * @date 2013-6-27

     * @version V1.0.0

     * @author linfenliang

     */

    void createTable(String tableName, String[] familys) throws IOException{

        HBaseAdmin admin = new HBaseAdmin(conf);//该类是管理HBase数据库表信息的接口，它提供对数据表的创建、删除、现实表现、设置表是否有效、添加删除表列族成员等

        if(admin.tableExists(tableName)){

            System.err.println(tableName + " already exists!");

        }else{

            HTableDescriptor tableDescr = new HTableDescriptor(tableName);//该类包含表的名字及其表的列族

            for(String familyName:familys){

                tableDescr.addFamily(new HColumnDescriptor(familyName));//向指定表中添加列族

                //HColumnDescriptor维护列族信息，如版本号、压缩设置。创建后不可修改，只能删除重建【注意测试数据对应的数据会被删掉】

                

//                tableDescr.setValue(key, value)//设置属性的值

//                tableDescr.getValue(key)//获取属性的值

//                tableDescr.getName()//获取表名称

            }

            admin.createTable(tableDescr);

            System.out.println("table "+ tableName +" is created !");

        }

    }

    /**

     * 删除表

     * @param tableName

     * @throws IOException

     * void

     * @date 2013-6-28

     * @version V1.0.0

     * @author linfenliang

     */

    void deleteTable(String tableName) throws IOException{

        HBaseAdmin admin = new HBaseAdmin(conf);

        admin.disableTable(tableName);

        admin.deleteTable(tableName);

        System.out.println("table "+ tableName +" is deleted !");

    }

    /**

     * 向表中添加记录

     * @param tableName

     * @param rowKey

     * @param family

     * @param qualifier

     * @param value

     * @throws IOException

     * void

     * @date 2013-6-28

     * @version V1.0.0

     * @author linfenliang

     */

    void addRecord(String tableName, String rowKey, String family, String qualifier, String value) throws IOException{

        HTable table = new HTable(conf,tableName);    //该表用来与HBase进行通信，但是HTable对更新操作时非线程安全的，我们实际使用中，

//        多为多线程通信机制，故此处不适合过多的线程处理，应使用HTablePool

//        HTablePool pool = new HTablePool();

//        pool.getTable(tableName).put(put);

//        pool.getTable(tableName).checkAndPut(row, family, qualifier, value, put)

        Put put = new Put(Bytes.toBytes(rowKey));

        put.add(Bytes.toBytes(family), Bytes.toBytes(qualifier), Bytes.toBytes(value));

        table.put(put);

        System.out.println("insertes recorded " + rowKey +" to table "+ tableName);

    }

    /**

     * 删除指定的记录

     * @param tableName

     * @param rowKey

     * @throws IOException

     * void

     * @date 2013-6-28

     * @version V1.0.0

     * @author linfenliang

     */

    void deleteRecord(String tableName, String rowKey) throws IOException{

        HTable table = new HTable(conf,tableName);

//        List<Delete> list = new ArrayList<Delete>();

//        Delete del = new Delete(rowKey.getBytes());

//        list.add(del);

//        table.delete(list);

        Delete delete = new Delete(rowKey.getBytes());

        table.delete(delete);

        System.out.println("record "+ rowKey+" is deleted");

    }

    /**

     * 删除多条记录

     * @param tableName

     * @param rowKeyList

     * @throws IOException

     * void

     * @date 2013-6-28

     * @version V1.0.0

     * @author linfenliang

     */

    void deleteRecords(String tableName, List<String> rowKeyList) throws IOException{

        HTable table = new HTable(conf,tableName);

        List<Delete> list = new ArrayList<Delete>();

        Delete del = null;

        for(String rowKey:rowKeyList){

            del = new Delete(rowKey.getBytes());

            list.add(del);

        }

        table.delete(list);

        System.out.println("record list "+ rowKeyList +" is deleted");

    }

    /**

     * 查询表中的记录

     * @param tableName

     * @param rowKey

     * @throws IOException

     * void

     * @date 2013-6-28

     * @version V1.0.0

     * @author linfenliang

     */

    void queryRecord(String tableName,String rowKey) throws IOException{

        HTable table = new HTable(conf,tableName);

        Get get = new Get(Bytes.toBytes(rowKey));

        Result rs =table.get(get);    //获取单行的值

        for(KeyValue kv:rs.raw()){

            System.out.println(String.format("ROW:%s,Family:%s,qualifier:%s,timestamp:%s,value:%s", new String(kv.getRow()),new String(kv.getFamily()),new String(kv.getQualifier()),kv.getTimestamp(),new String(kv.getValue())));

        }

    }

    /**

     * 查询表中所有记录

     * @param tableName

     * @throws IOException

     * void

     * @date 2013-6-28

     * @version V1.0.0

     * @author linfenliang

     */

    void queryAll(String tableName) throws IOException{

        HTable table = new HTable(conf,tableName);

        Scan scan = new Scan();

        ResultScanner rs = table.getScanner(scan); //

        for(Result result:rs){

            for(KeyValue kv:result.raw()){

                System.out.println(String.format("ROW:%s,Family:%s,qualifier:%s,timestamp:%s,value:%s", new String(kv.getRow()),new String(kv.getFamily()),new String(kv.getQualifier()),kv.getTimestamp(),new String(kv.getValue())));

            }

            System.out.println("------------------");

        }

    }

      

    

    public static void main(String[] args) throws Exception {

        String tableName = "V1_TRIP_DTL";

        String[] familys = {"id","name","score"};

        HBaseDemo demo = new HBaseDemo();

        System.out.println("-----创建表-----");

        demo.createTable(tableName, familys);

        System.out.println("-----添加记录-----");

        demo.addRecord(tableName, "2", "id", "", "1001");

        demo.addRecord(tableName, "2", "name", "", "linfenliang");

        demo.addRecord(tableName, "2", "score", "Math", "99");

        demo.addRecord(tableName, "2", "score", "english", "96");

        demo.addRecord(tableName, "2", "score", "Chinese", "98");

        System.out.println("-----查询记录-----");

        demo.queryRecord(tableName, "2");

        System.out.println("-----查询所有记录-----");

        demo.queryAll(tableName);

        System.out.println("-----删除记录-----");

        demo.deleteRecord(tableName, "2");

        System.out.println("-----删除后，查看记录-----");

        demo.queryAll(tableName);

        System.out.println("-----删除表-----");

        demo.deleteTable(tableName);

          

        

        

    }

 

}

以上为基本的增删查改操作。

 

  1. 与传统关系数据表结构比较

 

此为一个简单的示例：

设：有三张表：学生表（Student），课程表（Course），选课表（SC）

Student

字段

描述

S_NO

学号

S_NAME

姓名

S_SEX

性别

S_AGE

年龄

 

 

 

Course

字段

描述

C_NO

课程编号

C_NAME

课程名称

C_CREDIT

学分

 

SC

字段

描述

SC_SNO

学号

SC_CNO

课程编号

SC_SCORE

成绩

 

设学生信息表中有如下记录：

20131001,张三,男,18

课程表中有如下记录：

101,语文,4

选课表中有如下记录：

20131001,101,3

 

 

 

现在将其转换成Hbase中的存储结构

Student

Row Key

Column Family

Column Family

INFO

VALUE

Course

Value

20131001

INFO:S_NAME
INFO:S_SEX
INFO:S_AGE

张三
男
18

Course:C_NO
Course:C_NAME
Course:C_CREDIT

3
语文
4

 

 

Course

 

Row Key

Column Family

Column Family

INFO

VALUE

Student

Value

101

INFO:C_NAME
INFO:C_CREDIT

语文
4

Student:S_NO
Student:S_NAME
Student:S_SEX
Student:S_AGE

3x
张三
男
18

 

 

  1. 与当前应用的整合

下一步，我们对现有业务进行分析，并将Hbase应用到实际环境中

 

Oracle数据库中有张表插入次数特别多。即车辆行程明细表V_TRIP_DTL

还有一张表更新特别频繁，即车辆当前位置表，现在将这两张表迁移到Hbase

 

分析 V_TRIP_DTL

序号

字段名

数据类型

长度

允许空

字段说明

备注

1

ID

VARCHAR2

36

N

主键\外键：车辆信息表ID

关联表：

BS_VEHICLE _INFO

2

GatherTime

DATE

  

Y

采集时间

终端采集时间

3

ACCSTATUS

VARCHAR2

1

Y

ACC状态

ACC ON

ACC OFF

4

OBD_SPEED

NUMBER

5,2

Y

速度

单位（KM/H）

5

MAF

NUMBER

5,2

Y

空气流量

单位: Grams/sec

范围值：0~655.35

6

BASE_STATION

VARCHAR2

300

Y

基站信息(可能多个，基站数据用","分隔，多个基站信息之间用分号";"分隔)

基站编号,区域代码,国家代码,网络代码,信号强度; 基站编号,区域代码,国家代码,网络代码,信号强度;

7

FAULT_CODE

VARCHAR2

100

Y

故障码

多个之间用分号";"分隔

8

BABV

NUMBER

4,2

Y

  

电瓶电压

单位:V

范围值：0 ~65.53

网络用语（BABV）

9

MILEAGE

NUMBER

5,2

Y

里程

单位：m(米)

本条数据计算的里程

10

FUEL

NUMBER

5,2

Y

油耗

单位：L/s

本条数据计算的油耗；

网络用语（MGP）

11

G_FORCE

NUMBER

4,2

Y

G力值

计算急加速、急减速次数

12

PID_010C

NUMBER

7,2

Y

发动机转速

单位：rpm

范围值：0 ~16383.75

13

PID_0105

NUMBER

3

Y

冷却液温度(水温)

单位：°C

范围值：-40~215

14

PID_010B

NUMBER

3

Y

进气歧管绝对压力

单位：kPa

范围值：0~255

15

PID_0121

NUMBER

5

Y

故障灯亮行驶距离

单位: km

范围值：0 ~65535

16

PID_012F

NUMBER

3

Y

燃油液位输入

单位：%

范围值：0~100

17

PID_0131

NUMBER

5

Y

故障码清除以后行驶距离

单位: km

范围值：0 ~65535

18

PID_0146

NUMBER

3

Y

环境气温

单位：°C

范围值：-40~215

19

PID_015D

NUMBER

6,3

Y

喷油脉宽

单位: °C

范围值：-210.00~301.992

20

PID_010F

NUMBER

3

Y

进气温度

单位：°C

范围值：-40~215

21

PID_0104

NUMBER

3

Y

动力负荷计算值

单位：%

范围值：0~100

22

PID_0111

NUMBER

3

Y

节气门位置

单位：%

范围值：0~100

23

PID_0114

NUMBER

5,2

Y

氧传感器电压

  

24

PID_0122

NUMBER

8,2

Y

实际燃油压力(相对)

单位：Kpa

范围值：0~5177.265

25

PID_0123

NUMBER

5

Y

燃油实际压力

单位：Kpa

范围值：0 ~65535

26

PID_012D

NUMBER

5,2

Y

废气再循环误比

单位：%

范围值：-100 ~99.22

27

MILEAGE

NUMBER

9，2

Y

里程

单位：km, 对应车上里程表读数

28

OIL_MASS

NUMBER

5，2

Y

油量

单位：L，对应车上油量表读数

29

GPS_TIME

DATE

  

N

GPS时间

如：2012-11-20 12:32:12

30

RECEIVED_TIME

DATE

  

N

服务器接收时间

如：2012-11-20 12:32:12

31

REAL_LON

NUMBER

11，7

Y

真实经度

例如：120.11827

32

REAL_LAT

NUMBER

11，7

Y

真实纬度

例如：30.28269

33

OFFESET_LON

NUMBER

11，7

Y

偏转后经度

例如：120.11827

34

OFFESET_LAT

NUMBER

11，7

Y

偏转后纬度

例如：30.28269

35

CELL_LON

NUMBER

11，7

Y

基站经度

  

36

CELL_LAT

NUMBER

11，7

Y

基站纬度

  

37

GPS_SPEED

NUMBER

5,2

Y

速度

单位（km/h）

38

DIRECT

NUMBER

3

Y

方向

范围：0-359，正比为0，顺时针

39

ALTITUDE

NUMBER

4

Y

高度

单位（m）

40

ACCSTATUS

VARCHAR2

1

Y

ACC状态

ACC ON

ACC OFF

41

            

42

OIL_STATUS

VARCHAR2

1

Y

车辆油路状态

油路正常

油路断开

43

CIRCUIT_STATUS

VARCHAR2

1

Y

车辆电路状态

电路正常

电路断开

44

REMARK

VARCHAR2

100

Y

备注信息

  

 

转为Hbase的数据存储模式（部分数据未处理）：

Row Key

Column Family

Info

value

ID（UUID）

GatherTime

2013-6-28 16:55:10

accStatus

1

OBD_SPEED

  

MAF

  

BASE_STATION

  

FAULT_CODE

  

BABV

  

MILAGE

  

FUEL

  

G_FORCE

  

PID_010C

  

PID_010B

  

PID_010F

  

RECEIVED_TIME

  

…

  

 

 

建表、删表、数据插入，查询示例：

 

package com.sides.hbase.demo;

 

import java.io.IOException;

import java.text.DateFormat;

import java.text.SimpleDateFormat;

import java.util.ArrayList;

import java.util.Date;

import java.util.HashMap;

import java.util.List;

import java.util.Map;

import java.util.Map.Entry;

import java.util.UUID;

import java.util.concurrent.atomic.AtomicInteger;

 

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.hbase.HBaseConfiguration;

import org.apache.hadoop.hbase.HColumnDescriptor;

import org.apache.hadoop.hbase.HTableDescriptor;

import org.apache.hadoop.hbase.KeyValue;

import org.apache.hadoop.hbase.client.HBaseAdmin;

import org.apache.hadoop.hbase.client.HTableInterface;

import org.apache.hadoop.hbase.client.HTablePool;

import org.apache.hadoop.hbase.client.Put;

import org.apache.hadoop.hbase.client.Result;

import org.apache.hadoop.hbase.client.ResultScanner;

import org.apache.hadoop.hbase.client.Scan;

import org.apache.hadoop.hbase.util.Bytes;

 

/****************************************************************************

* com.sides.hbase.demo VehicleDetailInfoDao.java Created on 2013-6-28

* @Author: linfenliang

* @Description:

* @Version: 1.0

***************************************************************************/

public class VehicleDetailInfoDao extends Thread{

    public static final DateFormat dformat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    private static Configuration conf = null;

    private static HTablePool tablePool = null;

    static{

        Configuration hbaseConfig = new Configuration();

        hbaseConfig.set("hbase.zookeeper.quorum", "10.10.10.21");

//        hbaseConfig.set("hbase.zookeeper.property.clientPort", "2181");

        conf = HBaseConfiguration.create(hbaseConfig);

        

    }

    public static synchronized HTableInterface getHTable(String tableName){

        if(tablePool!=null){

            return tablePool.getTable(tableName);

        }else{

            tablePool = new HTablePool(conf,15);

            return tablePool.getTable(tableName);

        }

    }

    /**

     * 创建表

     * @param tableName

     * @param familys

     * @throws IOException

     * void

     * @date 2013-6-27

     * @version V1.0.0

     * @author linfenliang

     */

    void createTable(String tableName, String[] familys) throws IOException{

        HBaseAdmin admin = new HBaseAdmin(conf);

        if(admin.tableExists(tableName)){

            System.err.println(tableName + " already exists!");

        }else{

            HTableDescriptor tableDescr = new HTableDescriptor(tableName);

            for(String familyName:familys){

                tableDescr.addFamily(new HColumnDescriptor(familyName));

            }

            admin.createTable(tableDescr);

            System.out.println("table "+ tableName +" is created !");

        }

    }

    void deleteTable(String tableName) throws IOException{

        HBaseAdmin admin = new HBaseAdmin(conf);

        if(admin.tableExists(tableName)){

            admin.disableTable(tableName);

            admin.deleteTable(tableName);

        }else{

            System.out.println("table does not exists!");

        }

    }

    /**

     * 向表中添加记录

     * @param tableName

     * @param rowKey

     * @param family

     * @param qualifier

     * @param value

     * @throws IOException

     * void

     * @date 2013-6-28

     * @version V1.0.0

     * @author linfenliang

     */

    void addData(String tableName, String rowKey, String family, Map<String,String> map) throws IOException{

        HTableInterface table = getHTable(tableName);

        List<Put> putList = new ArrayList<Put>();

        for(Entry<String,String> entry:map.entrySet()){

            Put put = new Put(Bytes.toBytes(rowKey));

            put.add(Bytes.toBytes(family), Bytes.toBytes(entry.getKey()), Bytes.toBytes(entry.getValue()));

            putList.add(put);

        }

        table.put(putList);

        table.close();

        System.out.println("inserted record " + rowKey +" to table "+ tableName);

    }

    void getData(String tableName) throws IOException{

        HTableInterface table = getHTable(tableName);

        Scan scan = new Scan();

        long t0 = System.nanoTime();

        ResultScanner rs = table.getScanner(scan);

        long t1 = System.nanoTime();

        System.out.println("查询数据耗时:"+(t1-t0)/1000000.0);

        AtomicInteger a = new AtomicInteger();

        for(Result result:rs){

            a.incrementAndGet();

            System.out.println("Row Key:"+new String(result.getRow()));

            for(KeyValue kv:result.list()){

                System.out.println(String.format("Family:%s,qualifier:%s,timestamp:%s,value:%s", new String(kv.getFamily()),new String(kv.getQualifier()),kv.getTimestamp(),new String(kv.getValue())));

            }

            System.out.println("------------------");

        }

        System.out.println(a.get());

    }

    public static void main(String[] args) throws IOException {

        VehicleDetailInfoDao dao = new VehicleDetailInfoDao();

        String tableName = "V1_TRIP_DTL";

        String[] familys = new String[]{"INFO"};

//        dao.deleteTable(tableName);

        /*

         * 创建V1_TRIP_DTL表，列簇为INFO(存放名称)和VALUE(存放值)

         */

        dao.createTable(tableName, familys);

        Map<String,String> map = new HashMap<String, String>();

        map.put("GATHERTIME", dformat.format(new Date()));

        map.put("ACCSTATUS", "1");

        map.put("OBD_SPEED", "254.00");

        map.put("MAF", "655.35");

        map.put("BASE_STATION", "1823,4207,460,0,37;18431,4207,460,0,25;1822,4207,460,0,25;");

        map.put("FAULT_CODE", "9");

        map.put("BABV", "12.07");

        map.put("MILAGE", "70.56");

        map.put("FUEL", "63.34");

        map.put("G_FORCE", "0.00");

        map.put("PID_010C", "");

        map.put("PID_010B", "");

        map.put("PID_010F", "");

        map.put("RECEIVED_TIME", "2013-7-1 15:58:10");

        map.put("REMARK", "376");

        long t0 = System.nanoTime();

        for(int i=0;i<10000;i++){

            dao.addData(tableName, UUID.randomUUID().toString(), familys[0], map);

            System.out.println(i);

        }

        long t1 = System.nanoTime();

        System.out.println("插入1W数据耗时:"+(t1-t0)/1000000.0);

        dao.getData(tableName);

    }

}

 

  1. Hbase的不足

（1）目前存在问题：单线程插入速度不是很快，单独处理Hbase而不整合Hive的话，数据处理不方便，表的设计需要进一步优化（是否多列族，列族的设计），目前只是简单的测试了以下Hbase的可用性，方便学习。

（2）无法对行级数据操作，比如修改删除一条记录（Row）。

 

 

Hive学习总结[与Hbase整合]

  1. Hive的安装与运行

该部分参考杨文韬的《hive介绍及部署》一文进行整理

  1. Hive的安装

（1）下载并解压hive-0.11.0.tar.gz

 

tar xfz hive-0.11.0.tar.gz

 

目录结构如下图所示：

HBase与Hive入门研究

（2）修改配置文件

编辑hive-env.sh.template去掉后缀名

hive-env.sh文件中设置Hadoop_home路径：

 

HADOOP_HOME=/var/lib/tcommsvr/Hadoop_lin/hadoop-1.1.2

 

编辑hive-default.xml.template去掉后缀名

hive-site.xml文件中修改配置如下：

 

<property>

<name>hive.metastore.warehouse.dir</name>

<value>/var/lib/tcommsvr/Hadoop_lin/hive/warehouse</value>

<description>location of default database for the warehouse</description>

</property>

 

 

<property>

<name>hive.hwi.listen.host</name>

<value>10.10.10.21</value>

<description>This is the host address the Hive Web Interface will listen on</description>

</property>

 

 

<property>

<name>hive.aux.jars.path</name>

<value>file:///var/lib/tcommsvr/Hadoop_lin/hive-0.11.0/lib/hive-hbase-handler-0.11.0.jar,file:///var/lib/tcommsvr/Hadoop_lin/hive-0.11.0/lib/hbase-0.94.6.1.jar,file:///var/lib/tcommsvr/Hadoop_lin/hive-0.11.0/lib/zookeeper-3.4.3.jar</value>

</property>

  1. Hadoop的安装与处理

由于Hive的运行依赖于Hadoop的HDFS文件系统，故启动Hive之前需要安装Hadoop，

Hadoop安装目录：/var/lib/tcommsvr/Hadoop_lin/hadoop-1.1.2

HBase与Hive入门研究

修改配置文件core-side.xml

<?xml version="1.0"?>

<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>

 

<!-- Put site-specific property overrides in this file. -->

 

<configuration>

    <property>

        <name>fs.default.name</name>

        <value>hdfs://10.10.10.21:9000</value>

    </property>

</configuration>

 

修改配置文件hadoop-env.sh

 

export JAVA_HOME=/usr/lib/jvm/java-1.6.0-openjdk-1.6.0.0/jre

 

修改配置文件hdfs-site.xml

 

<?xml version="1.0"?>

<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>

 

<!-- Put site-specific property overrides in this file. -->

 

<configuration>

    <property>

        <name>dfs.replication</name>

        <value>1</value>

    </property>

</configuration>

 

修改配置文件mapred-site.xml

 

<?xml version="1.0"?>

<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>

 

<!-- Put site-specific property overrides in this file. -->

 

<configuration>

    <property>

        <name>mapred.job.tracker</name>    

        <value>10.10.10.21:9001</value>

    </property>

</configuration>

  1. Hive与Hbase整合处理

由于要整合Hbase与Hive，故

（1）拷贝Hbase中的hbase-site.xml到Hadoop的conf路径下

（2）拷贝hbase-0.94.6.1.jar和zookeeper-3.4.5.jar到hive/lib下

（3）拷贝hbase-0.94.6.1.jar到所有hadoop节点

不过如果Hadoop版本比较高，和Hadoop同步，则（2）（3）不是必须的，实际上，我在整合时没有做第二步和第三步

 

  1. Hive的运行

启动Hive前需要启动Hadoop

 

1.启动Hadoop：

cd /var/lib/tcommsvr/Hadoop_lin/hadoop-1.1.2/bin

sh start-all.sh

如果提示Permission denied则需要给所提示的脚本赋权，赋权命令：chmod u+x xxxx.sh

2.启动Hive：

cd /var/lib/tcommsvr/Hadoop_lin/hive-0.11.0/bin

（1）命令行模式：sh hive

该模式适用于采用命令行执行SQL

（2）服务器模式：sh hive --service hiveserver

该模式适用于采用编程方式执行SQL

（3）与Hbase整合模式：sh hive --service hiveserver -hiveconf hbase.master=master:60000

该模式为与Hbase整合后的编程模式，此模式下数据文件会保存在Hbase数据库中，采用Hbase模式存储数据，以下设计到的代码中，均采用改启动方式。

 

  1. Hive简介

  2. Hive概要分析

Hive是一个基于Hadoop文件系统之上的数据仓库架构。Hive定义了类SQL语言-Hive QL，它是一种类似的SQL语言。但只支持部分SQL，它所执行的一些条件查询其实质为调用Hadoop的Mapper和Reducer操作，由于Hive依赖于Hadoop（批处理操作系统），而Hadoop的任务具有较高的延迟性，任务的提交与处理都会有一定的延迟，这样，对于Hive，也是同样的具有较高的延迟性，我做了一个简单的测试，HBase数据库中有一张表，表中有三条数据，执行根据字段排序输入需要操作20s的时间。执行count操作，即便表中一条数据也没有，耗时也会在10s以上。由此可见，Hive的性能与oracle数据库比较，会很差劲。而且不提供在线事务处理，实时查询以及记录级（行级）的数据更新。但是在处理历史数据时（如车辆历史行程表），历史数据非常大的情况下，会比较好（未验证，不过可以通过分析得到：Hadoop适应于大数据的批量操作，Hive基于Hadoop，故执行大数据量的处理上，应该性能比较好，而且，估计在一定范围下，数据量越大，其性能会越好）。

 

  1. Hive的数据存储

HBase与Hive入门研究

 

该图实际为Hive的数据存储调用图或者称为数据交换图，Hive本身是不具备数据存储能力的，她的数据存储依赖于Hadoop文件系统。Hive也没有为数据建立索引，用户可以非常*的组织 Hive 中的表，只需要在创建表的时候告诉 Hive 数据中的列分隔符和行分隔符，Hive 就可以解析数据。Hive 中所有的数据都存储在 HDFS 中，Hive 中包含4种数据模型：表Table、外部表External Table、分区Partition、桶Bucket，下面逐一对这四种数据模型进行分析。

  1. 表Table

Hive 中的 Table 和数据库中的 Table 在概念上是类似的，每一个 Table 在 Hive 中都有一个相应的目录来存储数据（与Hbase是一致的）。例如，一个表 pvs，它在 HDFS 中的路径为：/wh/pvs，其中，wh 是在 hive-site.xml 中由 ${hive.metastore.warehouse.dir} 指定的数据仓库的目录，所有的 Table 数据（不包括 External Table）都保存在这个目录中（hadoop文件系统中的目录）。

  1. 外部表External Table

External Table 指向已经在 HDFS 中存在的数据，可以创建 Partition。它和 Table 在元数据的组织上是相同的，而实际数据的存储则有较大的差异。

Table 的创建过程和数据加载过程（这两个过程可以在同一个语句中完成），在加载数据的过程中，实际数据会被移动到数据仓库目录中；之后对数据对访问将会直接在数据仓库目录中完成。删除表时，表中的数据和元数据将会被同时删除。

External Table 只有一个过程，加载数据和创建表同时完成（CREATE EXTERNAL TABLE ……LOCATION），实际数据是存储在 LOCATION 后面指定的 HDFS 路径中，并不会移动到数据仓库目录中。当删除一个 External Table 时，仅删除元数据，表中的数据不会被删除。

  1. 分区Partition

Partition 对应于数据库中Partition 列的密集索引，但是 Hive 中 Partition 的组织方式与数据库中的很不相同。在 Hive 中，表中的一个 Partition 对应于表下的一个目录，所有的 Partition 数据都存储在对应的目录中。例如：pvs 表中包含 ds 和 city 两个 Partition，则对应于 ds = 20090801, city = US 的 HDFS 子目录为：/wh/pvs/ds=20090801/city=US；对应于 ds = 20090801, city = CA 的 HDFS 子目录为：/wh/pvs/ds=20090801/city=CA。

  1. 桶Bucket

Buckets 对指定列计算 hash，根据 hash 值切分数据，目的是为了便于并行，每一个 Buckets对应一个文件。将 user 列分散至 32 个Bucket上，首先对 user 列的值计算 hash，比如，对应 hash 值为 0 的 HDFS 目录为：/wh/pvs/ds=20090801/city=US/part-00000；对应hash 值为 20 的 HDFS 目录为：/wh/pvs/ds=20090801/city=US/part-00020。（不知所云）

  1. Hive的元数据相关介绍

  2. Hive的MetaStore作为元数据内容

包括表的名字，表的列和分区及其属性，表的属性（是否为外部表等），表的数据所在目录

Hive将元数据存储在RDBMS中，可以支持Derby,Mysql,Oracle,postgres,常用为Derby和Mysql（默认为Derby）

  1. Hive连接Derby数据库的三种模式：

Single User Mode 模式。利用此模式连接到一个In-Memory数据库Derby，一般用于单元测试。

Multi User Mode ，通过网络连接到一个数据库中，是最常用的模式（或者称为开发模式）

Remote Server Mode，用于非java客户端访问元数据库，在一个服务器端启动一个MetaStoreServer，在客户端利用Thrift协议通过MetaStoreServer访问元数据库（由于我们是基于java开发的，故不讨论此种模式）。

  1. 连接Derby

这种方式是最简单的存储方式，也是hive的默认配置，只需要在hive-default.xml或hive-site.xml做如下配置便可：

<!--表示使用derby存储-->

< property >

< name >javax.jdo.option.ConnectionURL</ name >

< value >jdbc:derby:;databaseName=metastore_db;create=true</ value >

</ property >

< property >

< name >javax.jdo.option.ConnectionDriverName</ name >

< value >org.apache.derby.jdbc.EmbeddedDriver</ value >

</ property >

<!—该方式属于本地方式，需要设置为true（metastroe服务与hive服务在一个进程中）-->

< property >

< name >hive.metastore.local</ name >

< value >true</ value >

</ property >

<!--（HDFS上的）数据目录-->

< property >

< name >hive.metastore.warehouse.dir</ name >

< value >/user/hive/warehouse</ value >

</ property >

使用derby存储方式时，运行hive会在当前目录生成一个derby文件和一个metastore_db目录。这种存储方式的弊端是在同一个目录下同时只能有一个hive客户端能使用数据库，否则会提示错误。

  1. 使用mysql存储元数据

需要将mysql的连接jar包拷贝到hive下面的lib下，然后修改hive-site.xml做如下配置：

<!—采用本地方式-->

< property >

< name >hive.metastore.local</ name >

< value >true</ value >

</ property >

<!—配置元数据存储为mysql方式-->

< property >

< name >javax.jdo.option.ConnectionURL</ name >

< value >jdbc:mysql://localhost/hive_remote?createDatabaseIfNotExist=true</ value >

</ property >

<!—mysql的驱动-->

< property >

< name >javax.jdo.option.ConnectionDriverName</ name >

< value >com.mysql.jdbc.Driver</ value >

</ property >

<!—连接mysql的用户名-->

< property >

< name >javax.jdo.option.ConnectionUserName</ name >

< value >root</ value >

</ property >

<!—连接mysql的密码-->

< property >

< name >javax.jdo.option.ConnectionPassword</ name >

< value >dandan</ value >

</ property >

 

  1. 使用远端mysql存储元数据

所谓的远程，并不是指mysql在其他服务器上，主要是meta服务是远程的，是否远程指的是metastore和hive服务是否在同一进程内，这种meta服务和hive服务可以分开，需要在Hive服务器启动meta服务，将mysql的连接jar拷贝到hive的安装目录下的lib下。修改配置文件hive-site.xml做如下配置：

服务器端：

< property >

< name >javax.jdo.option.ConnectionURL</ name >

< value >jdbc:mysql://localhost/hive_remote?createDatabaseIfNotExist=true</ value >

</ property >

< property >

< name >javax.jdo.option.ConnectionDriverName</ name >

< value >com.mysql.jdbc.Driver</ value >

</ property >

< property >

< name >javax.jdo.option.ConnectionUserName</ name >

< value >root</ value >

</ property >

< property >

< name >javax.jdo.option.ConnectionPassword</ name >

< value >dandan</ value >

</ property >

客户端：

<property>

<name>hive.metastore.local</name>

<value>false</value>

</property>

<property>

<name>hive.metastore.uris</name>

<value>thrift://ip:9083</value>

</property>

通过hive --service metastore启动meta服务

  1. Hive QL语言详解

  2. 数据定义DDL操作

创建内部表

hive> CREATE TABLE pokes (foo INT, bar STRING);

创建内部表并创建分区ds

hive> CREATE TABLE invites (foo INT, bar STRING) PARTITIONED BY (ds STRING);

创建外部表 (指定hdfs目录路径)

hive>create external table test(id int,name string) ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t' LOCATION '/test/hello′;

创建外部表 (指定本地目录路径)

hive>create external table test(id int,name string) ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t' LOCATION 'file:///home/test′;

显示所有表

hive> SHOW TABLES;

按正条件（正则表达式）显示表

hive> SHOW TABLES '.*s';

查看表结构字段

hive>desc (extended) test;

查看表结构，分隔符等信息

hive>desc formatted test;

分析查询语句

hive>explain select * from test;

表添加一列

hive> ALTER TABLE pokes ADD COLUMNS (new_col INT);

添加一列并增加列字段注释

hive> ALTER TABLE invites ADD COLUMNS (new_col2 INT COMMENT 'a comment');

修改表的列数据类型

hive> alter table ceshi change id id int; //将id数据类型修改为int 注意是两个id

更改表名

hive> ALTER TABLE events RENAME TO 3koobecaf;

删除列

hive> DROP TABLE pokes;

 

  1. 数据操作DML

将文件中的数据加载到表中

hive> LOAD DATA LOCAL INPATH './examples/files/kv1.txt' OVERWRITE INTO TABLE pokes;

加载本地数据，同时给定分区信息

hive> LOAD DATA LOCAL INPATH './examples/files/kv2.txt' OVERWRITE INTO TABLE invites PARTITION (ds='2008-08-15');

加载DFS数据 ，同时给定分区信息

hive> LOAD DATA INPATH '/user/myname/kv2.txt' OVERWRITE INTO TABLE invites PARTITION (ds='2008-08-15');

The above command will load data from an HDFS file/directory to the table. Note that loading data from HDFS will result in moving the file/directory. As a result, the operation is almost instantaneous.

 

  1. SQL操作

 

按条件查询

hive> SELECT a.foo FROM invites a WHERE a.ds='<DATE>';

将查询数据输出至目录

hive> INSERT OVERWRITE DIRECTORY '/tmp/hdfs_out' SELECT a.* FROM invites a WHERE a.ds='<DATE>';

将查询结果输出至本地目录

hive> INSERT OVERWRITE LOCAL DIRECTORY '/tmp/local_out' SELECT a.* FROM pokes a;

选择所有列到本地目录

hive> INSERT OVERWRITE TABLE events SELECT a.* FROM profiles a;

hive> INSERT OVERWRITE TABLE events SELECT a.* FROM profiles a WHERE a.key < 100;

hive> INSERT OVERWRITE LOCAL DIRECTORY '/tmp/reg_3' SELECT a.* FROM events a;

hive> INSERT OVERWRITE DIRECTORY '/tmp/reg_4' select a.invites, a.pokes FROM profiles a;

hive> INSERT OVERWRITE DIRECTORY '/tmp/reg_5' SELECT COUNT(1) FROM invites a WHERE a.ds='<DATE>';

hive> INSERT OVERWRITE DIRECTORY '/tmp/reg_5' SELECT a.foo, a.bar FROM invites a;

hive> INSERT OVERWRITE LOCAL DIRECTORY '/tmp/sum' SELECT SUM(a.pc) FROM pc1 a;

将一个表的统计结果插入另一个表中

hive> FROM invites a INSERT OVERWRITE TABLE events SELECT a.bar, count(1) WHERE a.foo > 0 GROUP BY a.bar;

hive> INSERT OVERWRITE TABLE events SELECT a.bar, count(1) FROM invites a WHERE a.foo > 0 GROUP BY a.bar;

JOIN

hive> FROM pokes t1 JOIN invites t2 ON (t1.bar = t2.bar) INSERT OVERWRITE TABLE events SELECT t1.bar, t1.foo, t2.foo;

将多表数据插入到同一表中

FROM src

INSERT OVERWRITE TABLE dest1 SELECT src.* WHERE src.key < 100

INSERT OVERWRITE TABLE dest2 SELECT src.key, src.value WHERE src.key >= 100 and src.key < 200

INSERT OVERWRITE TABLE dest3 PARTITION(ds='2008-04-08', hr='12') SELECT src.key WHERE src.key >= 200 and src.key < 300

INSERT OVERWRITE LOCAL DIRECTORY '/tmp/dest4.out' SELECT src.value WHERE src.key >= 300;

将文件流直接插入文件

hive> FROM invites a INSERT OVERWRITE TABLE events SELECT TRANSFORM(a.foo, a.bar) AS (oof, rab) USING '/bin/cat' WHERE a.ds > '2008-08-09';

This streams the data in the map phase through the script /bin/cat (like hadoop streaming). Similarly - streaming can be used on the reduce side (please see the Hive Tutorial or examples)

  1. 本地Hive与Hbase整合实例

前提条件：启动配置好的Hadoop与Hbase

启动Hive：sh hive --service hiveserver -hiveconf hbase.master=master:60000

 

VehicleDetailInfoDao.java

 

package com.sides.hive.demo;

 

import java.sql.Connection;

import java.sql.PreparedStatement;

import java.sql.ResultSet;

import java.sql.SQLException;

import java.util.Map;

 

/****************************************************************************

* com.sides.hive.demo VehicleDetailInfoDao.java Created on 2013-7-9

* @Author: linfenliang

* @Description:

* @Version: 1.0

***************************************************************************/

public class VehicleDetailInfoDao {

    void createTable(String tableName){

        String sql = "create table "+tableName+" (key string , value map<string,string>) stored by 'org.apache.hadoop.hive.hbase.HBaseStorageHandler' with serdeproperties (\"hbase.columns.mapping\" = \":key,info:\")";

        Connection conn = null;

        PreparedStatement pstmt = null;

        try {

            conn = DbConnectionPoolManage.getInstance().getConnection();

            pstmt = conn.prepareStatement(sql);

            pstmt.executeUpdate();

            System.out.println("表已创建成功" + tableName);

        } catch (SQLException e) {

            System.err.println(e);

        }finally{

            DbConnectionPoolManage.closeConnection(conn, pstmt, null);

        }

    }

    void dropTable(String tableName){

        String sql = "drop table " +tableName;

        Connection conn = null;

        PreparedStatement pstmt = null;

        try {

            conn = DbConnectionPoolManage.getInstance().getConnection();

            pstmt = conn.prepareStatement(sql);

            pstmt.executeUpdate();

            System.out.println("表已删除:" + tableName);

        } catch (SQLException e) {

            System.err.println(e);

        }finally{

            DbConnectionPoolManage.closeConnection(conn, pstmt, null);

        }

    }

    void showTables(){

        String sql = "show tables";

        Connection conn = null;

        PreparedStatement pstmt = null;

        ResultSet rs = null;

        try {

            conn = DbConnectionPoolManage.getInstance().getConnection();

            pstmt = conn.prepareStatement(sql);

            rs = pstmt.executeQuery();

            while(rs.next()){

                System.out.println(rs.getString(1));

            }

        } catch (SQLException e) {

            System.err.println(e);

        }finally{

            DbConnectionPoolManage.closeConnection(conn, pstmt, null);

        }

    }

    void descrTable(String tableName){

        String sql = "describe "+ tableName;

        Connection conn = null;

        PreparedStatement pstmt = null;

        ResultSet rs = null;

        try {

            conn = DbConnectionPoolManage.getInstance().getConnection();

            pstmt = conn.prepareStatement(sql);

            rs = pstmt.executeQuery();

            while(rs.next()){

                System.out.println(rs.getString(1)+"    "+rs.getString(2));

            }

        } catch (SQLException e) {

            System.err.println(e);

        }finally{

            DbConnectionPoolManage.closeConnection(conn, pstmt, null);

        }

    }

    

    void queryTable(String tableName){

        String sql = "select * from "+ tableName;

        Connection conn = null;

        PreparedStatement pstmt = null;

        ResultSet rs = null;

        try {

            conn = DbConnectionPoolManage.getInstance().getConnection();

            pstmt = conn.prepareStatement(sql);

            rs = pstmt.executeQuery();

            while(rs.next()){

                System.out.println(rs.getString(1)+"    "+rs.getObject(2));

            }

        } catch (SQLException e) {

            System.err.println(e);

        }finally{

            DbConnectionPoolManage.closeConnection(conn, pstmt, null);

        }

    }

    void insertRecord(String tableName, String key, Map<String,String> value){

        System.err.println("sorry, your job can not be supported");

    }

    void updateRecord(String tableName, String key, Map<String,String> value){

        System.err.println("sorry, your job can not be supported");

    }

    void deleteRecord(String tableName, String key){

        System.err.println("sorry, your job can not be supported");

    }

    /**

     * @param args

     * void

     * @date 2013-7-9

     * @version V1.0.0

     * @author linfenliang

     * @throws SQLException

     * @throws ClassNotFoundException

     */

    public static void main(String[] args) throws SQLException, ClassNotFoundException {

        VehicleDetailInfoDao dao = new VehicleDetailInfoDao();

//        dao.createTable("V11_TRIP_DTL");

        dao.showTables();

        dao.createTable("v_1");

        dao.showTables();

        dao.dropTable("v_1");

        dao.descrTable("hh_1406");

        dao.queryTable("hh_1406");

        

        

//        sh hive --service hiveserver -hiveconf hbase.master=master:60000

        

//        String filePath = "/var/lib/tcommsvr/Hadoop_lin/hive/lin.data.new";

//        String loadSql = "load data local inpath ' "+filePath+"' overwrite into table "+tableName;

//        System.out.println("Running："+loadSql);

//        res = stmt.executeQuery(loadSql);

        

        

//        String countSql = "select count(*) from " + tableName;

//        System.out.println("Running:"+countSql);

//        res = stmt.executeQuery(countSql);

//        while(res.next()){

//            System.out.println(res.getString(1));

//        }

        

//        String valueSql = "select value['name'] from " + tableName +" where value['name'] is not null";

//        System.out.println("Running:"+valueSql);

//        res = stmt.executeQuery(valueSql);

//        while(res.next()){

//            System.out.println(res.getString(1));

//        }

 

    }

 

}

 

在使用Hive操作数据库的时候，我使用了连接池（C3P0）

连接池配置如下：

DbConnectionPoolManage.java

 

package com.sides.hive.demo;

 

import java.beans.PropertyVetoException;

import java.sql.Connection;

import java.sql.ResultSet;

import java.sql.SQLException;

import java.sql.Statement;

 

import org.apache.log4j.Logger;

 

import com.mchange.v2.c3p0.ComboPooledDataSource;

 

 

 

public class DbConnectionPoolManage {

    private static Logger logger = Logger.getLogger(DbConnectionPoolManage.class);

    private static DbConnectionPoolManage instance;

    private ComboPooledDataSource dataSource;

 

     /**

* 初始加载C3P0相关配置参数

* @throws PropertyVetoException

*/

    private DbConnectionPoolManage() throws PropertyVetoException {

         dataSource = new ComboPooledDataSource();

        dataSource.setUser("");

        dataSource.setPassword("");

        dataSource.setJdbcUrl("jdbc:hive://119.255.194.51:10000/default");

        dataSource.setDriverClass("org.apache.hadoop.hive.jdbc.HiveDriver");

          

        

        //初始化时获取三个连接，取值应在minPoolSize与maxPoolSize之间。Default: 3 initialPoolSize

dataSource.setInitialPoolSize(3);

//连接池中保留的最大连接数。Default: 15 maxPoolSize

dataSource.setMaxPoolSize(15);

 

//// 连接池中保留的最小连接数。

dataSource.setMinPoolSize(1);

 

//当连接池中的连接耗尽的时候c3p0一次同时获取的连接数。Default: 3 acquireIncrement

dataSource.setAcquireIncrement(3);

 

    }

     /**

* 实例化DbConnectionPoolManage

*/

 

    public final static DbConnectionPoolManage getInstance() {

        if (instance == null) {

            synchronized (DbConnectionPoolManage.class) {

                try {

                    if (instance == null){

                        instance = new DbConnectionPoolManage();

                    }

                } catch (PropertyVetoException e) {

                    logger.error("初始C3P0连接池异常" + e);

                }

            }

        }

        return instance;

    }

 

    

    /**

     * notice:when can not get a connect , it will return null

     */

 

    public final synchronized Connection getConnection() {

        Connection con = null;

        try {

            con = dataSource.getConnection();

            return con;

        } catch (SQLException e) {

            logger.error("数据库连接失败" + e);

        }

 

        return con;

    }

    /**

     *

    

     */     

      

public static final void closeConnection( Connection conn,Statement stmt,ResultSet rs){

        closeResultSet(rs);

        closeStatement(stmt);

        closeConnection(conn);

}

private static final void closeConnection(Connection conn){

    if(conn!=null){

        try {

                conn.close();

            } catch (SQLException e) {

                logger.warn("close DataBase Connection failed , do not care");

            }finally{

                conn = null;

            }

    }

}

private static final void closeStatement(Statement stmt){

    if(stmt!=null){

        try {

            stmt.close();

            } catch (SQLException e) {

                logger.warn("close DataBase Statement failed , do not care");

            }finally{

                stmt = null;

            }

    }

}

private static final void closeResultSet(ResultSet rs){

    if(rs!=null){

        try {

            rs.close();

            } catch (SQLException e) {

                logger.warn("close DataBase ResultSet failed , do not care");

            }finally{

                rs = null;

            }

    }

}

}

输出运行结果：

 

13/07/16 09:12:17 INFO log.MLog: MLog clients using log4j logging.

13/07/16 09:12:18 INFO c3p0.C3P0Registry: Initializing c3p0-0.9.2.1 [built 20-March-2013 11:16:28 +0000; debug? true; trace: 10]

13/07/16 09:12:18 INFO impl.AbstractPoolBackedDataSource: Initializing c3p0 pool... com.mchange.v2.c3p0.ComboPooledDataSource [ acquireIncrement -> 3, acquireRetryAttempts -> 30, acquireRetryDelay -> 1000, autoCommitOnClose -> false, automaticTestTable -> null, breakAfterAcquireFailure -> false, checkoutTimeout -> 0, connectionCustomizerClassName -> null, connectionTesterClassName -> com.mchange.v2.c3p0.impl.DefaultConnectionTester, dataSourceName -> 1hge16q8v1rjrmfzmdsydm|4b222f, debugUnreturnedConnectionStackTraces -> false, description -> null, driverClass -> org.apache.hadoop.hive.jdbc.HiveDriver, factoryClassLocation -> null, forceIgnoreUnresolvedTransactions -> false, identityToken -> 1hge16q8v1rjrmfzmdsydm|4b222f, idleConnectionTestPeriod -> 0, initialPoolSize -> 3, jdbcUrl -> jdbc:hive://119.255.194.51:10000/default, maxAdministrativeTaskTime -> 0, maxConnectionAge -> 0, maxIdleTime -> 0, maxIdleTimeExcessConnections -> 0, maxPoolSize -> 15, maxStatements -> 0, maxStatementsPerConnection -> 0, minPoolSize -> 1, numHelperThreads -> 3, preferredTestQuery -> null, properties -> {user=******, password=******}, propertyCycle -> 0, statementCacheNumDeferredCloseThreads -> 0, testConnectionOnCheckin -> false, testConnectionOnCheckout -> false, unreturnedConnectionTimeout -> 0, userOverrides -> {}, usesTraditionalReflectiveProxies -> false ]

hh_1406

hive_hbase

hivehbase_test

test_0715

test_0716

test_lin

v10_trip_dtl

v3_trip_dtl

表已创建成功v_1

hh_1406

hive_hbase

hivehbase_test

test_0715

test_0716

test_lin

v10_trip_dtl

v3_trip_dtl

v_1

表已删除:v_1

key     int

value     map<string,string>

 

 

  1. 与传统SQL的比较

 

传统SQL与Hive-QL的比较：

HBase与Hive入门研究

 

  1. 查询语言

由于 SQL被广泛的应用在数据仓库中，因此，专门针对Hive 的特性设计了类 SQL 的查询语言HQL。熟悉SQL 开发的开发者可以很方便的使用 Hive 进行开发。

 

  1. 数据存储位置

Hive 是建立在 Hadoop之上的，所有 Hive 的数据都是存储在HDFS 中的。而数据库则可以将数据保存在块设备或者本地文件系统中。

 

  1. 数据格式

Hive 中没有定义专门的数据格式，数据格式可以由用户指定，用户定义数据格式需要指定三个属性：列分隔符（通常为空格、"\\t"、"\\x001″）、行分隔符（"\\n"）以及读取文件数据的方法（Hive中默认有三个文件格式 TextFile，SequenceFile 以及RCFile）。由于在加载数据的过程中，不需要从用户数据格式到 Hive 定义的数据格式的转换，因此，Hive 在加载的过程中不会对数据本身进行任何修改，而只是将数据内容复制或者移动到相应的 HDFS 目录中。而在数据库中，不同的数据库有不同的存储引擎，定义了自己的数据格式。所有数据都会按照一定的组织存储，因此，数据库加载数据的过程会比较耗时。

 

  1. 数据更新

由于 Hive 是针对数据仓库应用设计的，而数据仓库的内容是读多写少的。因此，Hive 中不支持对数据的改写和添加，所有的数据都是在加载的时候中确定好的。而数据库中的数据通常是需要经常进行修改的，因此可以使用 INSERT INTO ... VALUES 添加数据，使用 UPDATE ... SET 修改数据。

 

  1. 索引

Hive 在加载数据的过程中不会对数据进行任何处理，甚至不会对数据进行扫描，因此也没有对数据中的某些Key 建立索引。Hive 要访问数据中满足条件的特定值时，需要暴力扫描整个数据，因此访问延迟较高。由于MapReduce 的引入， Hive 可以并行访问数据，因此即使没有索引，对于大数据量的访问，Hive 仍然可以体现出优势。数据库中，通常会针对一个或者几个列建立索引，因此对于少量的特定条件的数据的访问，数据库可以有很高的效率，较低的延迟。由于数据的访问延迟较高，决定了Hive 不适合在线数据查询。

 

  1. 执行

Hive中大多数查询的执行是通过 Hadoop 提供的 MapReduce 来实现的（类似 select * from tbl 的查询不需要MapReduce）。而数据库通常有自己的执行引擎。

 

  1. 执行延迟

Hive 在查询数据的时候，由于没有索引，需要扫描整个表，因此延迟较高。另外一个导致 Hive 执行延迟高的因素是 MapReduce 框架。由于MapReduce 本身具有较高的延迟，因此在利用 MapReduce 执行 Hive 查询时，也会有较高的延迟。相对的，数据库的执行延迟较低。当然，这个低是有条件的，即数据规模较小，当数据规模大到超过数据库的处理能力的时候，Hive 的并行计算显然能体现出优势。

 

  1. 可扩展性

由于 Hive 是建立在 Hadoop 之上的，因此 Hive 的可扩展性是和Hadoop 的可扩展性是一致的（世界上最大的 Hadoop 集群在 Yahoo!，2009年的规模在 4000 台节点左右）。而数据库由于ACID 语义的严格限制，扩展行非常有限。目前最先进的并行数据库 Oracle 在理论上的扩展能力也只有 100 台左右。

 

  1. 数据规模

由于 Hive 建立在集群上并可以利用MapReduce 进行并行计算，因此可以支持很大规模的数据；对应的，数据库可以支持的数据规模较小。

 

  1. Hive的网络接口

在bin目录下输入：sh hive --service hwi

 

打开浏览器：

 

http://119.255.194.51:9998/hwi/

 

HBase与Hive入门研究

即可在浏览器中创建session，执行Hive-QL的操作。

由于该种方式不是主要研究方向，仅适合刚刚了解Hive时的辅助，故不再多叙。

标签：

相关文章