大数据数据库之HBase1.2.6详解大数据

$ ~/module/zookeeper-3.4.12/bin/zkServer.sh start

$ ~/module/hadoop-2.7.6/sbin/start-dfs.sh 
$ ~/module/hadoop-2.7.6/sbin/start-yarn.sh

[[email protected] hbase-1.2.6]$ tar zxvf hbase-1.2.6-bin.tar.gz -C /opt/module/

export JAVA_HOME=/opt/module/jdk1.8 
export HBASE_MANAGES_ZK=false

<configuration> 
<property> 
<name>hbase.rootdir</name> 
<value>hdfs://mycluster/hbase</value> 
</property> 
<property> 
<name>hbase.cluster.distributed</name> 
<value>true</value> 
</property> 
</property> 
<property> 
<name>hbase.zookeeper.quorum</name> 
<value>node21,node22,node23</value> 
</property> 
<property> 
<name>hbase.zookeeper.property.dataDir</name> 
<value>/opt/module/zookeeper-3.4.12/Data</value> 
</property> 
</configuration>

node21
node22
node23

node22

  rm -rf /opt/module/hbase-1.2.6/lib/hadoop-* 
$ rm -rf /opt/module/hbase-1.2.6/lib/zookeeper-3.4.6.jar

hadoop-annotations-2.7.6.jar  hadoop-mapreduce-client-app-2.7.6.jar     hadoop-mapreduce-client-hs-plugins-2.7.6.jar  
hadoop-auth-2.7.6.jar         hadoop-mapreduce-client-common-2.7.6.jar  hadoop-mapreduce-client-jobclient-2.7.6.jar   
hadoop-common-2.7.6.jar       hadoop-mapreduce-client-core-2.7.6.jar    hadoop-mapreduce-client-shuffle-2.7.6.jar      
hadoop-hdfs-2.7.6.jar         hadoop-mapreduce-client-hs-2.7.6.jar      hadoop-yarn-api-2.7.6.jar 
hadoop-yarn-client-2.7.6.jar  hadoop-yarn-common-2.7.6.jar              hadoop-yarn-server-common-2.7.6.jar
zookeeper-3.4.12.jar

$ find /opt/module/hadoop-2.7.6/ -name hadoop-annotations*

[[email protected] lib]$ ln -s /opt/module/hadoop-2.7.6/etc/hadoop/hdfs-site.xml /opt/module/hbase-1.2.6/conf/

$ scp -r /opt/module/hbase-1.2.6/ node22:/opt/module/ 
$ scp -r /opt/module/hbase-1.2.6/ node23:/opt/module/

$ bin/hbase-daemon.sh start master 
$ bin/hbase-daemon.sh start regionserver

<property> 
<name>hbase.master.maxclockskew</name> 
<value>180000</value> 
<description>Time difference of regionserver from master</description> 
</property>

$ bin/start-hbase.sh

$ bin/stop-hbase.sh

$ bin/hbase shell

hbase(main)> help

hbase(main)> list

hbase(main)> create 'student','info'

hbase(main) > put 'student','1001','info:name','Thomas' 
hbase(main) > put 'student','1001','info:sex','male'
hbase(main) > put 'student','1001','info:age','18' 
hbase(main) > put 'student','1002','info:name','Janna' 
hbase(main) > put 'student','1002','info:sex','female' 
hbase(main) > put 'student','1002','info:age','20'

hbase(main) > scan 'student' 
hbase(main) > scan 'student',{STARTROW => '1001', STOPROW    => '1001'}
hbase(main) > scan 'student',{STARTROW => '1001'}

hbase(main):012:0> describe ‘student’

hbase(main) > put 'student','1001','info:name','Nick' 
hbase(main) > put 'student','1001','info:age','100'

hbase(main) > get 'student','1001' 
hbase(main) > get 'student','1001','info:name'

hbase(main) > deleteall 'student','1001'

hbase(main) > delete 'student','1002','info:sex'

hbase(main) > truncate 'student'

hbase(main) > disable 'student'

hbase(main) > drop 'student'

hbase(main) > count 'student'

hbase(main) > alter 'student',{NAME=>'info',VERSIONS=>3}

$ tar -zxf ~/softwares/installations/apache-maven-3.5.0-bin.tar.gz -C ~/modules/

MAVEN_HOME=/home/admin/modules/apache-maven-3.5.0 
export PATH=$PATH:$MAVEN_HOME/bin

<dependency> 
<groupId>org.apache.hbase</groupId> 
<artifactId>hbase-server</artifactId> 
<version>1.3.1</version> 
</dependency> 
<dependency> 
<groupId>org.apache.hbase</groupId> 
<artifactId>hbase-client</artifactId> 
<version>1.3.1</version> 
</dependency> 
<dependency> 
<groupId>jdk.tools</groupId> 
<artifactId>jdk.tools</artifactId> 
<version>1.6</version> 
<scope>system</scope> 
<systemPath>${JAVA_HOME}/lib/tools.jar</systemPath> 
</dependency>

public static boolean isTableExist(String tableName) throws MasterNotRunningException, ZooKeeperConnectionException, IOException{ 
//在 HBase 中管理、访问表需要先创建 HBaseAdmin 对象 
//Connection connection = ConnectionFactory.createConnection(conf); 
//HBaseAdmin admin = (HBaseAdmin) connection.getAdmin(); HBaseAdmin admin = new HBaseAdmin(conf); 
return admin.tableExists(tableName); 
}

public static void createTable(String tableName, String... columnFamily) throws MasterNotRunningException, ZooKeeperConnectionException, IOException{ 
HBaseAdmin admin = new HBaseAdmin(conf); 
//判断表是否存在if(isTableExist(tableName)){ 
System.out.println("表" + tableName + "已存在"); 
//System.exit(0); 
}else{ 
//创建表属性对象,表名需要转字节 
HTableDescriptor descriptor = new HTableDescriptor(TableName.valueOf(tableName)); 
//创建多个列族 
for(String cf : columnFamily){ descriptor.addFamily(new HColumnDescriptor(cf)); 
} 
//根据对表的配置，创建表 
admin.createTable(descriptor); 
System.out.println("表" + tableName + "创建成功！"); 
  } 
}

public static void dropTable(String tableName) throws MasterNotRunningException, ZooKeeperConnectionException, IOException{ 
HBaseAdmin admin = new HBaseAdmin(conf); if(isTableExist(tableName)){ 
admin.disableTable(tableName); admin.deleteTable(tableName); 
System.out.println("表" + tableName + "删除成功！"); 
}else{ 
System.out.println("表" + tableName + "不存在！"); 
} 
}

public static void addRowData(String tableName, String rowKey, String columnFamily, String column, String value) throws IOException{ 
//创建 HTable 对象 
HTable hTable = new HTable(conf, tableName); 
//向表中插入数据 
Put put = new Put(Bytes.toBytes(rowKey)); 
//向 Put 对象中组装数据 
put.add(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(value)); 
hTable.put(put); hTable.close(); 
System.out.println("插入数据成功"); 
}

public static void deleteMultiRow(String tableName, String... rows) throws IOException{ HTable hTable = new HTable(conf, tableName); 
List<Delete> deleteList = new ArrayList<Delete>(); for(String row : rows){ 
Delete delete = new Delete(Bytes.toBytes(row)); deleteList.add(delete); 
} 
hTable.delete(deleteList); hTable.close(); 
}

public static void getRow(String tableName, String rowKey) throws IOException{ HTable table = new HTable(conf, tableName); 
Get get = new Get(Bytes.toBytes(rowKey)); 
//get.setMaxVersions();显示所有版本 
//get.setTimeStamp();显示指定时间戳的版本Result result = table.get(get); 
for(Cell cell : result.rawCells()){ 
System.out.println("行键:" + Bytes.toString(result.getRow())); 
System.out.println("列族" + Bytes.toString(CellUtil.cloneFamily(cell))); 
System.out.println("列:" + Bytes.toString(CellUtil.cloneQualifier(cell))); 
System.out.println("值:" + Bytes.toString(CellUtil.cloneValue(cell))); 
System.out.println("时间戳:" + cell.getTimestamp()); 
} 
}

public static void getRowQualifier(String tableName, String rowKey, String family, String qualifier) throws IOException{ 
HTable table = new HTable(conf, tableName); 
Get get = new Get(Bytes.toBytes(rowKey)); 
get.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier)); Result result = table.get(get); 
for(Cell cell : result.rawCells()){ 
System.out.println("行键:" + Bytes.toString(result.getRow())); 
System.out.println("列族" + Bytes.toString(CellUtil.cloneFamily(cell))); System.out.println("列:" + Bytes.toString(CellUtil.cloneQualifier(cell))); 
System.out.println("值:" + Bytes.toString(CellUtil.cloneValue(cell))); 
} 
}

$ bin/hbase mapredcp

$ export HBASE_HOME=/opt/module/hbase-1.2.6 
$ export HADOOP_HOME=/opt/module/hadoop-2.7.6 
$ export HADOOP_CLASSPATH=`${HBASE_HOME}/bin/hbase mapredcp`

$ ~/module/hadoop-2.7.6/bin/yarn jar lib/hbase-server-1.2.6.jar rowcounter student

hbase(main):001:0> create 'fruit','info'

$ ~/module/hadoop-2.7.6/bin/hdfs dfs -mkdir /input_fruit/ 
$ ~/module/hadoop-2.7.6/bin/hdfs dfs -put fruit.tsv /input_fruit/

$ ~/module/hadoop-2.7.6/bin/yarn jar lib/hbase-server-1.2.6.jar importtsv / 
-Dimporttsv.columns=HBASE_ROW_KEY,info:name,info:color fruit / hdfs://node21:8020/input_fruit

hbase(main):001:0> scan ‘fruit’

package com.z.hbase_mr; 
import java.io.IOException; 
import org.apache.hadoop.hbase.client.Put; 
import org.apache.hadoop.hbase.io.ImmutableBytesWritable; import org.apache.hadoop.hbase.mapreduce.TableReducer; import org.apache.hadoop.io.NullWritable; 
public class WriteFruitMRReducer extends TableReducer<ImmutableBytesWritable, Put, NullWritable> { 
@Override 
protected void reduce(ImmutableBytesWritable key, Iterable<Put> values, Context context) throws IOException, InterruptedException { 
//读出来的每一行数据写入到 fruit_mr 表中 
for(Put put: values){ context.write(NullWritable.get(), put); 
} 
} 
}

//组装 Job 
public int run(String[] args) throws Exception { 
//得到 Configuration 
Configuration conf = this.getConf(); 
//创建 Job 任务 
Job job = Job.getInstance(conf, this.getClass().getSimpleName()); 
job.setJarByClass(Fruit2FruitMRRunner.class); 
//配置 Job 
Scan scan = new Scan(); 
scan.setCacheBlocks(false); 
scan.setCaching(500); 
//设置 Mapper，注意导入的是 mapreduce 包下的，不是 mapred 包下的，后者是老版本 
TableMapReduceUtil.initTableMapperJob( 
"fruit", //数据源的表名 
scan, //scan 扫描控制器 
ReadFruitMapper.class,//设置 Mapper 类 
ImmutableBytesWritable.class,//设置 Mapper 输出 key 类型 
Put.class,//设置 Mapper 输出 value 值类型 
job//设置给哪个 JOB 
); 
//设置 Reducer 
TableMapReduceUtil.initTableReducerJob("fruit_mr", WriteFruitMRReducer.class,job); 
//设置 Reduce 数量，最少 1 个 
job.setNumReduceTasks(1); 
boolean isSuccess = job.waitForCompletion(true); 
if(!isSuccess){ 
throw new IOException("Job running with error"); 
} 
return isSuccess ? 0 : 1; 
}

public static void main( String[] args ) throws Exception{ 
Configuration conf = HBaseConfiguration.create(); 
int status = ToolRunner.run(conf, new Fruit2FruitMRRunner(), args);
System.exit(status);

$ ~/module/hadoop-2.7.6/bin/yarn jar ~/softwares/jars/hbase-0.0.1-SNAPSHOT.jar 
com.z.hbase.mr1.Fruit2FruitMRRunner

public static void main(String[] args) throws Exception { Configuration conf = HBaseConfiguration.create(); 
int status = ToolRunner.run(conf, new Txt2FruitRunner(), args); System.exit(status);

$ ~/module/hadoop-2.7.6/bin/yarn jar ~/softwares/jars/hbase-0.0.1-SNAPSHOT.jar 
com.z.hbase.mr2.Txt2FruitRunner

<property> 
<name>hive.zookeeper.quorum</name> 
<value>node21,node22,node23</value> 
<description>The list of ZooKeeper servers to talk to. This is only needed for read/write locks.</description> 
</property> 
<property> 
<name>hive.zookeeper.client.port</name> 
<value>2181</value> 
<description>The port of ZooKeeper servers to talk to. This is only needed for read/write locks.</description> 
</property>

CREATE TABLE emp( 
empno int,
ename string, 
job string,
mgr int, 
hiredate string, 
sal double, 
comm double, 
deptno int) 
row format delimited fields terminated by '/t';

hive> load data local inpath '/home/admin/softwares/data/emp.txt' into table emp;

hive> insert into table hive_hbase_emp_table select * from emp;

hive> select * from hive_hbase_emp_table;

hbase> scan ‘hbase_emp_table’

CREATE EXTERNAL TABLE relevance_hbase_emp(
empno int, 
ename string,
job string,
mgr int, 
hiredate string,
sal double,
comm double,
deptno int) 
STORED BY 
'org.apache.hadoop.hive.hbase.HBaseStorageHandler' WITH SERDEPROPERTIES ("hbase.columns.mapping" = 
":key,info:ename,info:job,info:mgr,info:hiredate,info:sal,info:comm,info:deptno") 
TBLPROPERTIES ("hbase.table.name" = "hbase_emp_table");

hive (default)> select * from relevance_hbase_emp;

export HBASE_HOME=/opt/module/hbase-1.2.6

INSERT INTO db_library.book (name, price) VALUES('Lie Sporting', '30'); 
INSERT INTO db_library.book (name, price) VALUES('Pride & Prejudice', '70'); 
INSERT INTO db_library.book (name, price) VALUES('Fall of Giants', '50');

$ bin/sqoop import / 
--connect jdbc:mysql://node21:3306/db_library / 
--username root / 
--password 123456 / 
--table book / 
--columns "id,name,price" / 
--column-family "info" / 
--hbase-create-table / 
--hbase-row-key "id" / 
--hbase-table "hbase_book" / 
--num-mappers 1 / 
--split-by id

hbase> create 'hbase_book','info'

hbase> scan ‘hbase_book’

hbase> status ‘node21’

hbase> whoami

hbase> list

hbase> count 'hbase_book'

hbase> describe 'hbase_book'

hbase> exist 'hbase_book'

hbase> is_enabled 'hbase_book' 
hbase> is_disabled 'hbase_book'

hbase> alter 'hbase_book', NAME => 'CF2', VERSIONS => 2

hbase> alter 'hbase_book', 'delete' => ’CF2’

hbase> disable 'hbase_book'

hbase> drop 'hbase_book'

hbase> delete ‘hbase_book’, ‘rowKey’, ‘CF:C’

hbase> truncate 'hbase_book'

hbase> create ‘table’, ‘cf’

hbase> create 't1', {NAME => 'f1'}, {NAME => 'f2'}, {NAME => 'f3'}

$ bin/hadoop distcp / 
hdfs://node21:8020/HbaseBackup/backup20180602 / 
hdfs://node21:8020/hbase

hbase> balance_switch false

hbase> hbase-daemon.sh stop regionserver

$ bin/graceful_stop.sh <RegionServer-hostname>

$ bin/stop-hbase.sh

$ touch conf/backup-masters

$ echo node22 > conf/backup-masters

$ scp -r conf/ node22:/opt/module/hbase-1.2.6/ 
$ scp -r conf/ node23:/opt/module/hbase-1.2.6/

0.98 版本之前：http://node21:60010 
0.98 版本之后：http://node21:16010

属性：dfs.namenode.handler.count 
解释：该属性是 NameNode 服务默认线程数，的默认值是 10，根据机器的可用内存可以调整为 50~100 
属性：dfs.datanode.handler.count 
解释：该属性默认值为 10，是 DataNode 的处理线程数，如果 HDFS 客户端程序读写请求比较多，可以调高到 15~20，设置的值越大，内存消耗越多，不要调整的过高，一般业务中， 
5~10 即可。

属性：dfs.replication 
解释：如果数据量巨大，且不是非常之重要，可以调整为 2~3，如果数据非常之重要，可以调整为 3~5。

属性：dfs.blocksize 
解释：块大小定义，该属性应该根据存储的大量的单个文件大小来设置，如果大量的单个文件都小于 100M，
建议设置成 64M 块大小，对于大于 100M 或者达到 GB 的这种情况，建议设置成 256M，一般设置范围波动在 64M~256M 之间。

属性：mapreduce.jobtracker.handler.count 
解释：该属性是 Job 任务线程数，默认值是 10，根据机器的可用内存可以调整为 50~100

属性：mapreduce.tasktracker.http.threads 
解释：定义 HTTP 服务器工作线程数，默认值为 40，对于大集群可以调整到 80~100

属性：mapreduce.task.io.sort.factor 
解释：文件排序时同时合并的数据流的数量，这也定义了同时打开文件的个数，默认值为 
10，如果调高该参数，可以明显减少磁盘 IO，即减少文件读取的次数。

属性：mapreduce.map.speculative 
解释：该属性可以设置任务是否可以并发执行，如果任务多而小，该属性设置为 true 可以明显加快任务执行效率，但是对于延迟非常高的任务，建议改为 false，这就类似于迅雷下载。

属性：mapreduce.map.output.compress、mapreduce.output.fileoutputformat.compress 
解释：对于大集群而言，建议设置 Map-Reduce 的输出为压缩的数据，而对于小集群，则不需要。

属性：mapreduce.tasktracker.map.tasks.maximum mapreduce.tasktracker.reduce.tasks.maximum 
解释：以上两个属性分别为一个单独的 Job 任务可以同时运行的 Map 和 Reduce 的数量。 
设置上面两个参数时，需要考虑 CPU 核数、磁盘和内存容量。假设一个 8 核的 CPU，业务内容非常消耗 CPU，那么可以设置 map 数量为 4，如果该业务不是特别消耗 CPU 类型的，
那么可以设置 map 数量为 40，reduce 数量为 20。这些参数的值修改完成之后，一定要观察是否有较长等待的任务，如果有的话，可以减少数量以加快任务执行，
如果设置一个很大的值，会引起大量的上下文切换，以及内存与磁盘之间的数据交换，这里没有标准的配置数值， 
需要根据业务和硬件配置以及经验来做出选择。 
在同一时刻，不要同时运行太多的 MapReduce，这样会消耗过多的内存，任务会执行的非常缓慢，我们需要根据 CPU 核数，内存容量设置一个 MR 任务并发的最大值，
使固定数据量的任务完全加载到内存中，避免频繁的内存和磁盘数据交换，从而降低磁盘 IO，提高性能。

$ sudo blockdev --setra 32768 /dev/sda

$ sudo sysctl -w vm.swappiness=0

$ ulimit -n 查看允许最大进程数 
$ ulimit -u 查看允许打开最大文件数

参数：zookeeper.session.timeout 
解 释 ：In hbase-site.xml, set zookeeper.session.timeout to 30 seconds or less to bound failure detection (20-30 seconds is a good start).
该值会直接关系到 master 发现服务器宕机的最大周期，默认值为 30 秒，如果该值过小，会在 HBase 在写入大量数据发生而 GC 时，导致RegionServer 短暂的不可用，
从而没有向 ZK 发送心跳包，最终导致认为从节点 shutdown。一般 20 台左右的集群需要配置 5 台 zookeeper。

create 'staff2','info','partition2',{NUMREGIONS => 15, SPLITALGO => 'HexStringSplit'}

aaaa 
bbbb 
cccc 
dddd

create 'staff3','partition3',SPLITS_FILE => 'splits.txt'

//自定义算法，产生一系列 Hash 散列值存储在二维数组中 
byte[][] splitKeys = 某个散列值函数 
//创建 HBaseAdmin 实例 
HBaseAdmin hAdmin = new HBaseAdmin(HBaseConfiguration.create()); 
//创建 HTableDescriptor 实例 
HTableDescriptor tableDesc = new HTableDescriptor(tableName); 
//通过 HTableDescriptor 实例和散列值二维数组创建带有预分区的 HBase 表 
hAdmin.createTable(tableDesc, splitKeys);

20170524000001 转成 10000042507102 
20170524000002 转成 20000042507102

20170524000001_a12e 
20170524000001_93i7

属性：dfs.support.append 
解释：开启 HDFS 追加同步，可以优秀的配合 HBase 的数据同步和持久化。默认值为 true.

属性：dfs.datanode.max.transfer.threads 
解释：HBase 一般都会同一时间操作大量的文件，根据集群的数量和规模以及数据动作，设置为 4096 或者更高。默认值：4096

属性：dfs.image.transfer.timeout 
解释：如果对于某一次数据操作来讲，延迟非常高，socket 需要等待更长的时间，建议把该值设置为更大的值（默认 60000 毫秒），以确保 socket 不会被 timeout 掉。

属性： 
mapreduce.map.output.compress mapreduce.map.output.compress.codec 
解释：开启这两个数据可以大大提高文件的写入效率，减少写入时间。第一个属性值修改为 
true，第二个属性值修改为：org.apache.hadoop.io.compress.GzipCodec 或者其他压缩方式

属性：dfs.datanode.failed.volumes.tolerated 
解释：默认为 0，意思是当 DataNode 中有一个磁盘出现故障，则会认为该 DataNode shutdown 了。
如果修改为 1，则一个磁盘出现故障时，数据会被复制到其他正常的 DataNode 上，当前的 DataNode 继续工作。

属性：hbase.regionserver.handler.count 
解释：默认值为 30，用于指定 RPC 监听的数量，可以根据客户端的请求数进行调整，读写请求较多时，增加此值。

属性：hbase.hregion.max.filesize 
解释：默认值 10737418240（10GB），如果需要运行 HBase 的 MR 任务，可以减小此值， 因为一个 region 对应一个 map 任务,
如果单个 region 过大，会导致 map 任务执行时间过长。该值的意思就是，如果 HFile 的大小达到这个数值，则这个 region 会被切分为两个 Hfile。

属性：hbase.client.write.buffer 
解释：用于指定 HBase 客户端缓存，增大该值可以减少 RPC 调用次数，但是会消耗更多内存，反之则反之。一般我们需要设定一定的缓存大小，以达到减少 RPC 次数的目的。

属性：hbase.client.scanner.caching 
解释：用于指定 scan.next 方法获取的默认行数，值越大，消耗内存越大。

hbase.regionserver.global.memstore.upperLimit：0.4 
hbase.regionserver.global.memstore.lowerLimit：0.38

hbase(main):002:0> create_namespace 'ns_school'

hbase(main):004:0> create 'ns_school:tbl_student','info'

//获取配置 conf 
private Configuration conf = HBaseConfiguration.create(); 
//微博内容表的表名 
private static final byte[] TABLE_CONTENT = Bytes.toBytes("ns_weibo:content"); 
//用户关系表的表名 
private static final byte[] TABLE_RELATION = Bytes.toBytes("ns_weibo:relation"); 
//微博收件箱表的表名 
private static final byte[] TABLE_INBOX = Bytes.toBytes("ns_weibo:inbox"); 
/** 
*初始化命名空间 
[email protected] args 
*/ 
public void initNamespace(){ HBaseAdmin admin = null; try { 
Connection connection = ConnectionFactory.createConnection(conf); admin = (HBaseAdmin) connection.getAdmin(); 
//命名空间类似于关系型数据库中的 schema，可以想象成文件夹 
NamespaceDescriptor weibo = NamespaceDescriptor 
.create("ns_weibo") 
.addConfiguration("creator", "Jinji") 
.addConfiguration("create_time", System.currentTimeMillis() + "") 
.build(); admin.createNamespace(weibo); 
} catch (MasterNotRunningException e) { 
e.printStackTrace(); 
} catch (ZooKeeperConnectionException e) { e.printStackTrace(); 
} catch (IOException e) { e.printStackTrace(); 
}finally{ 
if(null != admin){ try { 
admin.close(); 
} catch (IOException e) { e.printStackTrace(); 
}

/** 
*创建微博内容表 
*Table Name:ns_weibo:content 
*RowKey:用户 ID_时间戳 
*ColumnFamily:info 
*ColumnLabel:标题,内容,图片 URL 
*Version:1 个版本 
*/ 
public void createTableContent(){ HBaseAdmin admin = null; 
Connection connection = null; try { 
connection = ConnectionFactory.createConnection(conf); admin = (HBaseAdmin) connection.getAdmin(); 
//创建表表述 
HTableDescriptor contentTableDescriptor = new 
HTableDescriptor(TableName.valueOf(TABLE_CONTENT)); 
//创建列族描述 
HColumnDescriptor infoColumnDescriptor = new HColumnDescriptor(Bytes.toBytes("info")); 
//设置块缓存 
infoColumnDescriptor.setBlockCacheEnabled(true); 
//设置块缓存大小 
infoColumnDescriptor.setBlocksize(2097152); 
//设置压缩方式 
// infoColumnDescriptor.setCompressionType(Algorithm.SNAPPY); 
//设置版本确界 
infoColumnDescriptor.setMaxVersions(1); 
infoColumnDescriptor.setMinVersions(1); contentTableDescriptor.addFamily(infoColumnDescriptor); 
admin.createTable(contentTableDescriptor); 
} catch (IOException e) { e.printStackTrace(); 
} finally{ 
if(null != admin){ try { 
admin.close(); connection.close(); 
} catch (IOException e) { e.printStackTrace(); 
} 
} 
} 
}

/** 
*用户关系表 
*Table Name:ns_weibo:relation 
*RowKey:用户 ID 
*ColumnFamily:attends,fans 
*ColumnLabel:关注用户 ID，粉丝用户 ID 
*ColumnValue:用户 ID 
*Version：1 个版本 
*/ 
public void createTableRelation(){ 
HBaseAdmin admin = null; try { 
Connection connection = ConnectionFactory.createConnection(conf); admin = (HBaseAdmin) connection.getAdmin(); 
HTableDescriptor relationTableDescriptor = new HTableDescriptor(TableName.valueOf(TABLE_RELATION)); 
//关注的人的列族 
HColumnDescriptor attendColumnDescriptor = new HColumnDescriptor(Bytes.toBytes("attends")); 
//设置块缓存 
attendColumnDescriptor.setBlockCacheEnabled(true); 
//设置块缓存大小 
attendColumnDescriptor.setBlocksize(2097152); 
//设置压缩方式 
//attendColumnDescriptor.setCompressionType(Algorithm.SNAPPY); 
//设置版本确界 
attendColumnDescriptor.setMaxVersions(1); 
attendColumnDescriptor.setMinVersions(1); 
//粉丝列族 
HColumnDescriptor fansColumnDescriptor = new 
HColumnDescriptor(Bytes.toBytes("fans"));
fansColumnDescriptor.setBlockCacheEnabled(true); 
fansColumnDescriptor.setBlocksize(2097152); 
fansColumnDescriptor.setMaxVersions(1); 
fansColumnDescriptor.setMinVersions(1); 
relationTableDescriptor.addFamily(attendColumnDescriptor); 
relationTableDescriptor.addFamily(fansColumnDescriptor);
 admin.createTable(relationTableDescriptor); 
} catch (MasterNotRunningException e) { e.printStackTrace(); 
} catch (ZooKeeperConnectionException e) { e.printStackTrace(); 
} catch (IOException e) { e.printStackTrace(); 
}finally{ 
if(null != admin){ try { 
admin.close(); 
} catch (IOException e) 
{ e.printStackTrace(); 
}
}

/** 
*创建微博收件箱表 
*Table Name: ns_weibo:inbox 
*RowKey:用户 ID 
*ColumnFamily:info 
*ColumnLabel:用户 ID_发布微博的人的用户 ID 
*ColumnValue:关注的人的微博的 RowKey 
*Version:1000 
*/ 
public void createTableInbox(){ HBaseAdmin admin = null; try { 
Connection connection = ConnectionFactory.createConnection(conf); admin = (HBaseAdmin) connection.getAdmin(); 
HTableDescriptor inboxTableDescriptor = new HTableDescriptor(TableName.valueOf(TABLE_INBOX)); 
HColumnDescriptor infoColumnDescriptor = new 
HColumnDescriptor(Bytes.toBytes("info")); 
infoColumnDescriptor.setBlockCacheEnabled(true);
infoColumnDescriptor.setBlocksize(2097152); 
infoColumnDescriptor.setMaxVersions(1000);
infoColumnDescriptor.setMinVersions(1000); 
inboxTableDescriptor.addFamily(infoColumnDescriptor);
admin.createTable(inboxTableDescriptor); 
} catch (MasterNotRunningException e) { e.printStackTrace(); 
} catch (ZooKeeperConnectionException e) { e.printStackTrace(); 
} catch (IOException e) { e.printStackTrace(); 
}finally{ 
if(null != admin){ try { 
admin.close(); 
} catch (IOException e) { e.printStackTrace(); 
}

package com.z.hbase.weibo; 
 
public class Message { private String uid; 
private String timestamp; private String content; 
public String getUid() { 
return uid; 
} 
public void setUid(String uid) { this.uid = uid; 
} 
public String getTimestamp() { 
return timestamp; 
} 
public void setTimestamp(String timestamp) { this.timestamp = timestamp; 
} 
public String getContent() { 
return content; 
} 
public void setContent(String content) { this.content = content; 
} 
@Override 
public String toString() { 
return "Message [uid=" + uid + ", timestamp=" + timestamp + ", content=" + content +"]"; 
} 
}

/** 
*发布微博 
*a、微博内容表中数据+1 
*b、向微博收件箱表中加入微博的 Rowkey 
*/ 
public void publishContent(String uid, String content){ Connection connection = null; 
try { 
connection = ConnectionFactory.createConnection(conf); 
//a、微博内容表中添加 1 条数据，首先获取微博内容表描述 
Table contentTable = connection.getTable(TableName.valueOf(TABLE_CONTENT)); 
//组装 Rowkey 
long timestamp = System.currentTimeMillis(); String rowKey = uid + "_" + timestamp; 
//添加微博内容 
Put put = new Put(Bytes.toBytes(rowKey)); put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("content"), timestamp, 
Bytes.toBytes(content)); 
contentTable.put(put); 
//b、向微博收件箱表中加入发布的 Rowkey 
//b.1、查询用户关系表，得到当前用户有哪些粉丝 
Table relationTable = connection.getTable(TableName.valueOf(TABLE_RELATION)); 
//b.2、取出目标数据 
Get get = new Get(Bytes.toBytes(uid)); get.addFamily(Bytes.toBytes("fans")); 
Result result = relationTable.get(get); 
List<byte[]> fans = new ArrayList<byte[]>(); 
//遍历取出当前发布微博的用户的所有粉丝数据for(Cell cell : result.rawCells()){ 
fans.add(CellUtil.cloneQualifier(cell)); 
} 
//如果该用户没有粉丝，则直接 return 
if(fans.size() <= 0) return; 
//开始操作收件箱表 
Table inboxTable = connection.getTable(TableName.valueOf(TABLE_INBOX)); 
//每一个粉丝，都要向收件箱中添加该微博的内容，所以每一个粉丝都是一个 Put 
对象 
List<Put> puts = new  
ArrayList<Put>(); for(byte[] fan : fans){ 
Put fansPut = new Put(fan); 
fansPut.addColumn(Bytes.toBytes("info"), Bytes.toBytes(uid), timestamp, 
Bytes.toBytes(rowKey)); 
puts.add(fansPut); 
} 
inboxTable.put(puts); 
} catch (IOException e) { 
e.printStackTrace(); 
}finally{ 
if(null != connection){ try { 
connection.close(); 
} catch (IOException e) { e.printStackTrace(); 
} 
}

/** 
*关注用户逻辑 
*a、在微博用户关系表中，对当前主动操作的用户添加新的关注的好友 
*b、在微博用户关系表中，对被关注的用户添加粉丝（当前操作的用户） 
*c、当前操作用户的微博收件箱添加所关注的用户发布的微博 rowkey 
*/ 
public void addAttends(String uid, String... attends){ 
//参数过滤 
if(attends == null || attends.length <= 0 || uid == null || uid.length() <= 0){ 
return; 
} 
Connection connection = null; try { 
connection = ConnectionFactory.createConnection(conf); 
//用户关系表操作对象（连接到用户关系表） 
Table relationTable = connection.getTable(TableName.valueOf(TABLE_RELATION)); 
List<Put> puts = new ArrayList<Put>(); 
//a、在微博用户关系表中，添加新关注的好友 
Put attendPut = new Put(Bytes.toBytes(uid)); for(String attend : attends){ 
//为当前用户添加关注的人 
attendPut.addColumn(Bytes.toBytes("attends"), Bytes.toBytes(attend), 
Bytes.toBytes(attend)); 
//b、为被关注的人，添加粉丝 
Put fansPut = new Put(Bytes.toBytes(attend)); fansPut.addColumn(Bytes.toBytes("fans"), Bytes.toBytes(uid), 
Bytes.toBytes(uid)); 
//将所有关注的人一个一个的添加到 puts（List）集合中 
puts.add(fansPut); 
} 
puts.add(attendPut); relationTable.put(puts); 
//c.1、微博收件箱添加关注的用户发布的微博内容（content）的 rowkey 
Table contentTable = connection.getTable(TableName.valueOf(TABLE_CONTENT)); Scan scan = new Scan(); 
//用于存放取出来的关注的人所发布的微博的 rowkey 
List<byte[]> rowkeys = new ArrayList<byte[]>(); 
for(String attend : attends){ 
//过滤扫描 rowkey，即：前置位匹配被关注的人的 uid_ 
RowFilter filter = new RowFilter(CompareFilter.CompareOp.EQUAL, new 
SubstringComparator(attend + "_")); 
//为扫描对象指定过滤规则 
scan.setFilter(filter); 
//通过扫描对象得到 scanner 
ResultScanner result = contentTable.getScanner(scan); 
//迭代器遍历扫描出来的结果集 
Iterator<Result> iterator = result.iterator(); while(iterator.hasNext()){ 
//取出每一个符合扫描结果的那一行数据 
Result r = iterator.next(); 
for(Cell cell : r.rawCells()){ 
//将得到的 rowkey 放置于集合容器中 
rowkeys.add(CellUtil.cloneRow(cell)); 
} 
} 
} 
//c.2、将取出的微博 rowkey 放置于当前操作的用户的收件箱中 
if(rowkeys.size() <= 0) return; 
//得到微博收件箱表的操作对象 
Table inboxTable = connection.getTable(TableName.valueOf(TABLE_INBOX)); 
/用于存放多个关注的用户的发布的多条微博 rowkey 信息 
List<Put> inboxPutList = new ArrayList<Put>(); 
for(byte[] rk : rowkeys){ 
Put put = new Put(Bytes.toBytes(uid)); 
//uid_timestamp 
String rowKey = Bytes.toString(rk); 
//截取 uid 
String attendUID = rowKey.substring(0, rowKey.indexOf("_")); 
long timestamp = Long.parseLong(rowKey.substring(rowKey.indexOf("_") + 1)); 
//将微博 rowkey 添加到指定单元格中 
put.addColumn(Bytes.toBytes("info"), Bytes.toBytes(attendUID), timestamp, rk); inboxPutList.add(put); 
} 
inboxTable.put(inboxPutList); 
} catch (IOException e) { 
e.printStackTrace(); 
}finally{ 
if(null != connection){ try { 
connection.close(); 
} catch (IOException e) { 
// TODO Auto-generated catch block e.printStackTrace(); 
} 
} 
} 
}

/** 
*取消关注（remove) 
*a、在微博用户关系表中，对当前主动操作的用户删除对应取关的好友 
*b、在微博用户关系表中，对被取消关注的人删除粉丝（当前操作人） 
*c、从收件箱中，删除取关的人的微博的 rowkey 
* 
*/ 
public void removeAttends(String uid, String... attends){ 
//过滤数据 
if(uid == null || uid.length() <= 0 || attends == null || attends.length <= 0) return; 
try { 
Connection connection = ConnectionFactory.createConnection(conf); 
//a、在微博用户关系表中，删除已关注的好友 
Table relationTable = connection.getTable(TableName.valueOf(TABLE_RELATION)); 
//待删除的用户关系表中的所有数据 List<Delete> deleteList = new ArrayList<Delete>(); 
//当前取关操作者的 uid 对应的 Delete 对象 
Delete attendDelete = new Delete(Bytes.toBytes(uid)); 
//遍历取关，同时每次取关都要将被取关的人的粉丝-1 for(String attend : attends){ 
attendDelete.addColumn(Bytes.toBytes("attends"), Bytes.toBytes(attend)); 
//b、在微博用户关系表中，对被取消关注的人删除粉丝（当前操作人） Delete fansDelete = new Delete(Bytes.toBytes(attend)); 
fansDelete.addColumn(Bytes.toBytes("fans"), Bytes.toBytes(uid)); deleteList.add(fansDelete); 
} 
deleteList.add(attendDelete); relationTable.delete(deleteList); 
//c、删除取关的人的微博 rowkey 从 收件箱表中 
Table inboxTable = connection.getTable(TableName.valueOf(TABLE_INBOX)); 
Delete inboxDelete = new Delete(Bytes.toBytes(uid)); for(String attend : attends){ 
inboxDelete.addColumn(Bytes.toBytes("info"), Bytes.toBytes(attend)); 
} 
inboxTable.delete(inboxDelete) 
} catch (IOException e) { e.printStackTrace(); 
} 
}

/** 
* 获取微博实际内容 
*a、从微博收件箱中获取所有关注的人的发布的微博的 rowkey 
*b、根据得到的 rowkey 去微博内容表中得到数据 
*c、将得到的数据封装到 Message 对象中 
*/ 
public List<Message> getAttendsContent(String uid){ Connection connection = null; 
try { 
connection = ConnectionFactory.createConnection(conf) 
Table inboxTable = connection.getTable(TableName.valueOf(TABLE_INBOX)); 
//a、从收件箱中取得微博 rowKey 
Get get = new Get(Bytes.toBytes(uid)); 
//设置最大版本号 
get.setMaxVersions(5); 
List<byte[]> rowkeys = new ArrayList<byte[]>(); Result result = inboxTable.get(get); 
for(Cell cell : result.rawCells()){ rowkeys.add(CellUtil.cloneValue(cell)); 
} 
//b、根据取出的所有 rowkey 去微博内容表中检索数据 
Table contentTable = connection.getTable(TableName.valueOf(TABLE_CONTENT)); List<Get> gets = new ArrayList<Get>(); 
//根据 rowkey 取出对应微博的具体内容 
for(byte[] rk : rowkeys){ 
Get g = new Get(rk); gets.add(g); 
} 
//得到所有的微博内容的 result 对象 
Result[] results = contentTable.get(gets); 
//将每一条微博内容都封装为消息对象 
List<Message> messages = new ArrayList<Message>(); for(Result res : results){ 
for(Cell cell : res.rawCells()){ 
Message message = new Message(); 
String rowKey = Bytes.toString(CellUtil.cloneRow(cell)); 
String userid = rowKey.substring(0, rowKey.indexOf("_")); String timestamp = rowKey.substring(rowKey.indexOf("_") + 1); 
String content = Bytes.toString(CellUtil.cloneValue(cell)); 
message.setContent(content); message.setTimestamp(timestamp); 
message.setUid(userid); 
messages.add(message) 
} 
} 
return messages; 
} catch (IOException e) { 
e.printStackTrace(); 
}finally{ 
try { 
connection.close(); 
} catch (IOException e) { 
e.printStackTrace(); 
} 
} 
return null; 
}

/** 
*发布微博内容 
*添加关注 
*取消关注 
*展示内容 
*/ 
public void testPublishContent(WeiBo wb){ 
wb.publishContent("0001", "今天买了一包空气，送了点薯片，非常开心！！"); wb.publishContent("0001", "今天天气不错。"); 
} 
public void testAddAttend(WeiBo wb){ wb.publishContent("0008", "准备下课！"); 
wb.publishContent("0009", "准备关机！"); wb.addAttends("0001", "0008", "0009"); 
} 
public void testRemoveAttend(WeiBo wb){ wb.removeAttends("0001", "0008"); 
} 
public void testShowMessage(WeiBo wb){ 
List<Message> messages = wb.getAttendsContent("0001"); for(Message message : messages){ 
System.out.println(message); 
} 
} 
public static void main(String[] args) { 
WeiBo weibo = new WeiBo(); 
weibo.initTable(); 
weibo.testPublishContent(weibo); 
weibo.testAddAttend(weibo);
weibo.testShowMessage(weibo);
weibo.testRemoveAttend(weibo); 
weibo.testShowMessage(weibo); 
}