分布式存储系统-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);
    }

}

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

扫码关注云+社区

领取腾讯云代金券