分布式存储系统-HBASE

简介

HBase –Hadoop Database,是一个高可靠性、高性能、面向列、可伸缩的分布式存储系统，利用HBse技术可在廉价PC Server上搭建起大规模结构化存储集群。HBase利用Hadoop HDFS作为文件存储系统，利用Hadoop MapReduce来处理HBase中的海量数据，利用Zookeeper作为协调工具。

HBase(NoSQL)的数据模型

HBase 存储的网站页面数据示例

下面讲下关于HBase的相关名词和概念：

表(table)，是存储管理数据的。

行键(row key)，类似于MySQL中的主键。Mysql中的主键可有可无。

行键是HBase表天然自带的。

列族（colum family）,列的集合。

HBASE列族是需要在定义表时指定的，只需要定义列族而不需要定义列。列是在插入数据时动态增加的。

时间戳（timestamp），是64位长整形整数，在我们数据库里它是一种日期类型可以精确到毫秒级别。在HBASE，是列（也称作标签、修饰符）的一个属性。

行健和列确定的单元格，可以存储多个数据，每个数据还有时间戳属性。数据具有版本特性。

如果不指定时间戳或者版本，默认取最新的数据。

HDFS中存储的数据都是字节数组。

HBASE中表中的数据是按照行健的顺序物理存储的。MySQL中是按照插入的顺序存储的。行健是按照ASSIC码排序的。

HBASE 是面向列的数据库，按照列族进行存储。关系型数据库是按照行存储的。

列必须用族来定义，任意一列有如下形式：族：标签。族和标签都可以是任意的字符串。

物理上将同族数据存储在一起。

数据可通过时间戳来区分版本。

HBase(NoSQL)的物理模型

HBase架构图

Hbase适合海量数据（如20PB）的秒级简单查询的数据库。

Hbase 表中的记录按照行健拆分成一个一个的region。

Region 存储 region server（单独的物理机器）。这样，对表的操作转化为对多台region server 的并行查询。

Hbase的体系结构：

Hbase是主从式结构，HMaster、HRegionServer

Hmaster:

Hbase允许有多个HMaster.通过Zookeeper Election机制保证总有一个Master运行。

Master为regionserver分配region，负责region server的负载均衡，发现失效的region server 并重新分配其上的region

HRegionServer:

Regionserver维护master分配给它的region,处理对这些region的IO请求，负责切分在运行过程中变得过大的region。

Zookeeper:

保证任何时候 ，集群中只有一个running master

存储所有 region的寻址入口

实时监控region server的状态，将region server的上线和下线信息，实时通知给master

存储 hbase的schema,包括哪些table每个table有哪些column family

Client访问hbase上数据的过程并不需要master参与，寻址访问zookeeper 和 region server,数据读写访问regionserver。HRegionServer主要负责响应用户I/O请求,向HDFS文件系统中读写数据，是HBase中最核心的模块。

Hbase中有两张特殊的Table,-ROOT-和.META.

.META.:记录了用户的region信息，.META .可以有多个region

-ROOT-:记录了.META.表的region信息，-ROOT-只有一个region, Zookeeper中记录了-ROOT-表的location。

Client访问用户数据之前需要首先访问zookeeper,然后访问-ROOT-表，接着访问.META.表，最后才能找到用户数据的位置访问。

HBase伪分布安装：

1. 解压缩、重命名、设置环境变量

2. 修改$HBASE_HOME/conf/hbase-env.sh，修改内容如下：

   export JAVA_HOME=/usr/local/jdk
   export HBASE_MANAGES_ZK=true

3. 修改$HBASE_HOME/conf/hbase-site.xml，修改内容如下：

   <property>
     <name>hbase.rootdir</name>
     <value>hdfs://hadoop:9000/hbase</value>
   </property>
   <property>
     <name>hbase.cluster.distributed</name>
     <value>true</value>
   </property>
   <property>
     <name>hbase.zookeeper.quorum</name>
     <value>hadoop</value>
   </property>
   <property>
     <name>dfs.replication</name>
     <value>1</value>
   </property>

4. (可选)文件regionservers的内容改为hadoop

5. 启动hbase，执行命令start-hbase.sh
   
   启动hbase之前，确保hadoop是运行正常的，并且可以写入文件

6. 验证：(1)执行jps，发现新增加了3个java进程，分别是HMaster、HRegionServer、HQuorumPeer，证明安装成功。
   
   (2)使用浏览器访问http://hadoop:60010 查看webUI.

HBase集群搭建

1.hbase的集群搭建过程(在原来的hadoop上的hbase伪分布基础上进行搭建)

  1.1 集群结构，主节点(hmaster)是hadoop，从节点(region server)是hadoop1和hadoop2

   1.2 修改hadoop上的hbase的几个文件

   (1)修改hbase-env.sh的最后一行export HBASE_MANAGES_ZK=false

(2)修改hbase-site.xml文件的hbase.zookeeper.quorum的值为hadoop0,hadoop1,hadoop2。修改dfs.replicationg改为3，表示副本数为3。

  (3)修改regionservers文件(存放的region server的hostname)，内容修改为hadoop1、hadoop2

  1.3 复制hadoop中的hbase文件夹到hadoop1、hadoop2中

复制hadoop中的/etc/profile到hadoop1、hadoop2中，在hadoop1、hadoop2上执行source /etc/profile

  1.4 首先启动hadoop，然后启动zookeeper集群在各个节点上分别启动。

最后在hadoop上启动hbase集群。

出现的问题及解决办法：

因为在搭建集群的时候，hbase在运行状态下，修改了配置造成：

1） hbase web 界面进不去

2） hbase shell 操作没有结果

3） 运行stop-hbase.sh 关闭不了

解决办法：

1） 把所有节点关掉，再次启动

2） 还有一个解决办法：杀死hbase进程 kill -9 pid 用ps –ef | grep 查看hbase进程

下面我们学习下HBase下的shell操作。

HBase Shell

1. 进入hbase shell，终端：

   hbase shell

2. hbase shell操作：
   
   创建表： create ‘表名称’, ‘列族名称1’,’列族名称2’,’列族名称N’
   
   添加记录： put ‘表名称’, ‘行名称’, ‘列名称:’, ‘值’
   
   查看记录: get ‘表名称’, ‘行名称’
   
   查看表中的记录总数: count ‘表名称’
   
   删除一张表: 先要屏蔽该表，才能对该表进行删除。
   
   第一步 disable ‘表名称’ 第二步 drop ‘表名称’
   
   查看所有记录: scan “表名称”
   
   查看某个表某个列中所有数据: scan “表名称” {COLUMNS=>’列族名称:列名称’}
   
   更新记录: 就是重写一遍进行覆盖

eg:

表users,有三个列族user_id,address,info:

创建表:

create 'users','user_id','address','info'

列出全部表

 list

得到表的描述

 describe 'users'

删除表：

disable  ‘user’
drop ‘user

添加记录：

put 'users','xiaoming','info:age','24'

获取一条记录

1.取得一个id的所有数据

get 'users','xiaoming'

2.获取一个id，一个列族的所有数据

get 'users','xiaoming','info' 

3.获取一个id，一个列族中一个列的 所有数据

get 'users','xiaoming','info:age'

更新记录

put 'users','xiaoming','info:age' ,'29' 

get 'users','xiaoming','info:age' 

put 'users','xiaoming','info:age' ,'30' 

get 'users','xiaoming','info:age' 

获取单元格数据的版本数据

get 'users','xiaoming',{COLUMN=>'info:age',VERSIONS=>1} 

get 'users','xiaoming',{COLUMN=>'info:age',VERSIONS=>2} 

get 'users','xiaoming',{COLUMN=>'info:age',VERSIONS=>3} 

获取单元格数据的某个版本数据

get 'users','xiaoming',{COLUMN=>'info:age',TIMESTAMP=>1364874937056} 

全表扫描

scan 'users'

行数的判断是根据行健来判断的

HBase 中查找数据的方式：

1. get方式

2. scan方式

删除xiaoming值的’info:age’字段

delete 'users','xiaoming','info:age' 

get 'users','xiaoming' 

HBase列：put动态增加 delete动态减少

删除整行

deleteall 'users','xiaoming'

统计表的行数

 count 'users'

清空表

 truncate 'users' 

坚持下，下面我们看下关于Java针对HBase的操作，这个时候java程序员应该兴奋才是！！！

HBase的Java_API操作

package hbase;
import java.io.IOException;

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.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;

/*
 * 创建表、插入记录、查询一条记录、遍历所有记录、删除表
 * */
public class HBaseApp {
    private static final String TABLE_NAME = "table1";
    private static final String FAMILY_NAME = "family1";
    private static final String  ROW_KEY = "rowkey1";

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

    Configuration conf  = HBaseConfiguration.create();
    conf.set("hbase.rootdir", "hdfs://hadoop:9000/hbase");
    //使用eclipse时必须添加这个，否则无法定位
    conf.set("hbase.zookeeper.quorum", "hadoop");
    //创建表、删除表使用HBaseAdmin
   final HBaseAdmin hBaseAdmin = new HBaseAdmin(conf);
   createTable(hBaseAdmin);
   //deleteTable(hBaseAdmin);

    //插入记录、查询一条记录、遍历所有记录使用HTable
     final HTable hTable = new HTable(conf, TABLE_NAME);
//     putRecord(hTable);
//     getRecord(hTable);

      scanTable(hTable);
    }

    private static void scanTable(final HTable hTable) throws IOException {
        Scan scan = new Scan();
         final ResultScanner scanner = hTable.getScanner(scan);

         for (Result result : scanner) {
             final byte[] value = result.getValue(FAMILY_NAME.getBytes(),"age".getBytes());
             System.out.println(result + "\t"+ new String(value));
        }
    }

    //查询所有记录

    //查询一条记录
    private static void getRecord(final HTable hTable) throws IOException {
         Get get = new Get(ROW_KEY.getBytes());
         final Result result = hTable.get(get);
         final byte[] value = result.getValue(FAMILY_NAME.getBytes(),"age".getBytes());

         System.out.println(result + "\t"+ new String(value));
    }

      //插入一条记录
    private static void putRecord(final HTable hTable) throws IOException {
        Put put = new Put(ROW_KEY.getBytes());
         put.add(FAMILY_NAME.getBytes(), "age".getBytes(), "25".getBytes());
         hTable.put(put);
         hTable.close();
    }

    //删除表
    private static void deleteTable(final HBaseAdmin hBaseAdmin)
            throws IOException {
          hBaseAdmin.disableTable(TABLE_NAME);
          hBaseAdmin.deleteTable(TABLE_NAME);
    }

    //创建表
    private static void createTable(final HBaseAdmin hBaseAdmin)
            throws IOException {
        if(!hBaseAdmin.tableExists(TABLE_NAME)){
//表名
HTableDescriptor descriptor = new HTableDescriptor(TABLE_NAME);
//添加列族
HColumnDescriptor family = new HColumnDescriptor(FAMILY_NAME);
descriptor.addFamily(family);
hBaseAdmin.createTable(descriptor);
           }
    }
}

2>:
hdfs数据批量导入到hbase中

create 'wlan_log', 'cf'
package hbase;

import java.text.SimpleDateFormat;
import java.util.Date;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.mapreduce.TableOutputFormat;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;

public class BatchImport {
    static class BatchImportMapper extends Mapper<LongWritable, Text, LongWritable, Text>{
        SimpleDateFormat dateformat1=new SimpleDateFormat("yyyyMMddHHmmss");
        Text v2 = new Text();

        protected void map(LongWritable key, Text value, Context context) throws java.io.IOException ,InterruptedException {
            final String[] splited = value.toString().split("\t");
            try {
                final Date date = new Date(Long.parseLong(splited[0].trim()));
                final String dateFormat = dateformat1.format(date);
                String rowKey = splited[1]+":"+dateFormat;
                v2.set(rowKey+"\t"+value.toString());
                context.write(key, v2);
            } catch (NumberFormatException e) {
                final Counter counter = context.getCounter("BatchImport", "ErrorFormat");
                counter.increment(1L);
                System.out.println("出错了"+splited[0]+" "+e.getMessage());
            }
        };
    }

    static class BatchImportReducer extends TableReducer<LongWritable, Text, NullWritable>{
        protected void reduce(LongWritable key, java.lang.Iterable<Text> values,    Context context) throws java.io.IOException ,InterruptedException {
            for (Text text : values) {
                final String[] splited = text.toString().split("\t");

                final Put put = new Put(Bytes.toBytes(splited[0]));
                put.add(Bytes.toBytes("cf"), Bytes.toBytes("date"), Bytes.toBytes(splited[1]));
                put.add(Bytes.toBytes("cf"), Bytes.toBytes("msisdn"), Bytes.toBytes(splited[2]));
                //省略其他字段，调用put.add(....)即可
                context.write(NullWritable.get(), put);
            }
        };
    }

    public static void main(String[] args) throws Exception {
        final Configuration configuration = new Configuration();
        //设置zookeeper
        configuration.set("hbase.zookeeper.quorum", "hadoop");
        //设置hbase表名称
        configuration.set(TableOutputFormat.OUTPUT_TABLE, "wlan_log");
        //将该值改大，防止hbase超时退出
        configuration.set("dfs.socket.timeout", "180000");

        final Job job = new Job(configuration, "HBaseBatchImport");

        job.setMapperClass(BatchImportMapper.class);
        job.setReducerClass(BatchImportReducer.class);
        //设置map的输出，不设置reduce的输出类型
        job.setMapOutputKeyClass(LongWritable.class);
        job.setMapOutputValueClass(Text.class);

        job.setInputFormatClass(TextInputFormat.class);
        //不再设置输出路径，而是设置输出格式类型
        job.setOutputFormatClass(TableOutputFormat.class);

        FileInputFormat.setInputPaths(job, "hdfs://hadoop:9000/input");

        job.waitForCompletion(true);
    }

}