HBase分布式数据库技术与应用 课件12项目七-任务一-基于HBase的微博案例

《HBase分布式数据库实战》拓展篇HBase分布式数据库实战篇章引入章节任务HBase分布式数据库实战项目七

基于HBase的微博案例任务一

基于HBase的微博案例回顾知识点1

sqoop介绍回顾回顾知识点2sqoop版本回顾知识点3sqoop安装回顾知识点4sqoop基本命令任务引入对基于Web的应用，还有一种很重要的数据，即用户交互数据。这一类数据包含了用户的访问网站的行为习惯。通过分析用户交互数据，就可以获取用户在网站上的活动信息。Facebook为此创建了一个名为FacebookInsight的系统，该系统需要一个可扩展的存储系统。基于HBase，Facebook可以很方便地横向扩展服务规模，提供给数百万用户，也可以继续使用他们已有的运行大规模HBase机群的经验。该系统每天处理数百亿条事件，记录数百个参数。任务概要任务描述:本任务讲解一个HBase典型案例，通过编写HBaseAPI程序来模拟微博运行过程，可以实现发布微博、关注/取关用户、查看用户微博内容等基本功能。任务教学目标：掌握常用的HBaseJavaAPI功能及用法。掌握HBaseAPI开发流程。一数据表设计二项目架构教学内容三定义常量四HBaseUtil封装五HBaseDao封装六测试代码数据表设计知识点一（一）数据表设计设计一个列族info，包含一个列Content，用于存放发布的微博内容。行键设计为用户id+时间戳，每行数据表示某个用户在某一时刻发布的微博内容。所有用户发布的所有微博内容均存储在一张表中微博内容表（一）数据表设计设计两个列族attends、fans，表示当前用户的关注者和粉丝。每个列族可以有很多列，每个列表示一个具体的用户，列的值与列名保持相同，仅起到标识用户的作用。行键设计为用户id。如果某个用户没有关注任何其他用户，则该行数据的attends列族为空；同样地，如果没有粉丝，则fans列族为空用户关系表（一）数据表设计设计为一个列族info，可以包括多个列，分别表示当前用户关注的其他用户。行键设计为用户id，每行数据表示当前用户关注的其他用户所发布的微博内容，每列的值设置3个版本，内容为关注者所发布的微博内容的行键（uid+ts）。即列的值是第一张表（微博内容表）的外键微博收件箱表（一）数据表设计数据示例知识点二项目架构（二）项目架构<dependencies><dependency><groupId>org.apache.hbase</groupId><artifactId>hbase-server</artifactId><version>1.4.13</version></dependency><dependency><groupId>org.apache.hbase</groupId><artifactId>hbase-client</artifactId><version>1.4.13</version></dependency></dependencies>创建Maven工程，在pom.xml文件中添加项目所需要的依赖，如左所示。在src/main/resources文件夹下添加配置文件hbase-site.xml，其中包含项目要连接的集群信息，这样就不用在代码里再设置了。在src/main/java下创建4个Java包：constants、utils、dao、test，分别表示常量、工具类、数据操作、测试。定义常量知识点三（三）定义常量在constants包下创建Constants类，存放项目需要用到的常量。包括HBase配置信息、命名空间、微博内容表及其列族信息、用户关系表及其列族信息、收件箱表及其列族信息。下面是关键代码。定义常量（三）定义常量//HBase配置信息publicstaticfinalConfigurationCONFIGURATION=HBaseConfiguration.create();//命名空间publicstaticfinalStringNAMESPACE="weibo";//微博内容表publicstaticfinalStringCONTENT_TABLE="weibo:content";publicstaticfinalStringCONTENT_TABLE_CF="info";publicstaticfinalintCONTENT_TABLE_VERSIONs=1;（三）定义常量//用户关系表publicstaticfinalStringRELATION_TABLE="weibo:relation";publicstaticfinalStringRELATION_TABLE_CF1="attends";publicstaticfinalStringRELATION_TABLE_CF2="fans";publicstaticfinalintRELATION_TABLE_VERSIONs=1;//收件箱表publicstaticfinalStringINBOX_TABLE="weibo:inbox";publicstaticfinalStringINBOX_TABLE_CF="info";publicstaticfinalintINBOX_TABLE_VERSIONs=2;HBaseUtil封装知识点四（四）HBaseUtil封装在utils包下创建HBaseUtil类，提供项目通用的功能，包括创建命名空间、判断表是否存在、创建表。下面是关键代码。HBaseUtil封装（四）HBaseUtil封装publicstaticvoidcreateNamespace(StringnameSpace)throwsIOException{//1获取Connection对象Connectionconnection=ConnectionFactory.createConnection(Constants.CONFIGURATION);//2获取Admin对象Adminadmin=connection.getAdmin();//3构建命名空间描述器NamespaceDescriptornamespaceDescriptor=NamespaceDescriptor.create(nameSpace).build();//4创建命名空间admin.createNamespace(namespaceDescriptor);//5关闭资源admin.close();connection.close();}1.创建命名空间（四）HBaseUtil封装privatestaticbooleanisTableExist(StringtableName)throwsIOException{//1获取Connection对象Connectionconnection=ConnectionFactory.createConnection(Constants.CONFIGURATION);//2获取Admin对象Adminadmin=connection.getAdmin();//3判断是否存在booleanexists=admin.tableExists(TableName.valueOf(tableName));//4关闭资源admin.close();connection.close();//5返回结果returnexists;}2.判断表是否存在（四）HBaseUtil封装publicstaticvoidcreateTable(StringtableName,intversions,String...cfs)throwsIOException{//1判断是否传入了列族信息if(cfs.length<=0){System.out.println("请传入列族信息");return;}//2判断表是否存在if(isTableExist(tableName)){System.out.println("表"+tableName+"已存在");return;}//3获取Connection对象Connectionconnection=ConnectionFactory.createConnection(Constants.CONFIGURATION);3.创建表//4获取Admin对象Adminadmin=connection.getAdmin();//5创建表描述器HTableDescriptorhTableDescriptor=newHTableDescriptor(TableName.valueOf(tableName));//6添加列族信息for(Stringcf:cfs){HColumnDescriptorhColumnDescriptor=newHColumnDescriptor(cf);//7设置版本ColumnDescriptor.setMaxVersions(versions);hTableDescriptor.addFamily(hColumnDescriptor);}//8创建表对象admin.createTable(hTableDescriptor);//9关闭资源admin.close();connection.close();}}HBaseDao封装知识点五（五）HBaseDao封装在dao包下创建HBaseDao类，实现与业务相关的数据操作功能。难点是要同时操作3张表，要处理3张表之间的联系。某个uid用户发布了一条微博，系统做了这些事情：在微博内容表中，添加一条数据；在用户关系表中，查找uid的fans用户；在微博收件箱表中，为所有fans用户更新当前uid用户的微博发布信息。HBaseDao封装（五）HBaseDao封装微博内容表中，行键设计为uid+时间戳。发布微博TablecontTable=connection.getTable(TableName.valueOf(Constants.CONTENT_TABLE));longts=System.currentTimeMillis();StringrowKey=uid+"_"+ts;PutcontPut=newPut(Bytes.toBytes(rowKey));contPut.addColumn(Bytes.toBytes(Constants.CONTENT_TABLE_CF),Bytes.toBytes("content"),Bytes.toBytes(content));contTable.put(contPut);（五）HBaseDao封装在用户关系表中，查找当前uid的fans信息。遍历每一个粉丝，在微博收件箱表中为每个粉丝更新uid用户的微博发布信息。构建Get对象，行键为uid，列族为fans，传入用户关系表中，返回一个result对象，其中包含uid的所有粉丝信息。发布微博TablerelaTable=connection.getTable(TableName.valueOf(Constants.RELATION_TABLE));GetrelaGet=newGet(Bytes.toBytes(uid));relaGet.addFamily(Bytes.toBytes(Constants.RELATION_TABLE_CF2));Resultresult=relaTable.get(relaGet);（五）HBaseDao封装解析result对象，获取每一个cell对象，每个cell对象包含列族fans中的各列的列名及列的值。将获取到的粉丝信息作为新的行键，构建微博收件箱表的Put对象，列族为info，列名为当前用户的uid，值是当前新添加微博的行键（uid+时间戳）。每一个Put对象存放在创建的ArrayList列表中。循环遍历每一个粉丝，进行如上操作。发布微博ArrayList<Put>inboxPuts=newArrayList<>();for(Cellcell:result.rawCells()){PutinboxPut=newPut(CellUtil.cloneQualifier(cell));inboxPut.addColumn(Bytes.toBytes(Constants.INBOX_TABLE_CF),Bytes.toBytes(uid),Bytes.toBytes(rowKey));inboxPuts.add(inboxPut);}（五）HBaseDao封装如果当前uid用户没有粉丝（ArrayList列表为空），则不对微博收件箱表做任何操作。否则，获取微博收件箱表对象，并将ArrayList列表中的Put对象传入到微博收件箱表中。发布微博//判断当前uid是否有粉丝if(inboxPuts.size()>0){TableinboxTable=connection.getTable(TableName.valueOf(Constants.INBOX_TABLE));inboxTable.put(inboxPuts);inboxTable.close();}（五）HBaseDao封装关注用户同样需要操作3张表。在用户关系表中，当前uid用户在attends列族中添加若干个用户，同时这些用户在fans列族中添加当前uid用户。此外，在微博收件箱表中，以当前uid用户为行键，在列族info中添加若干个列，每列表示一个关注人，每列的值存放此关注人发布的微博的行键（版本存放2条）。关注用户（五）HBaseDao封装添加uid用户的关注人，且同步更新关注人的粉丝信息关注用户TablerelaTable=connection.getTable(TableName.valueOf(Constants.RELATION_TABLE));PutattendPut=newPut(Bytes.toBytes(uid));ArrayList<Put>relaPuts=newArrayList<>();for(Stringattend:attends){//对于每一个attend对象，要同时做两件事情attendPut.addColumn(Bytes.toBytes(Constants.RELATION_TABLE_CF1),Bytes.toBytes(attend),Bytes.toBytes(attend));（五）HBaseDao封装添加uid用户的关注人，且同步更新关注人的粉丝信息关注用户PutfanPut=newPut(Bytes.toBytes(attend));fanPut.addColumn(Bytes.toBytes(Constants.RELATION_TABLE_CF2),Bytes.toBytes(uid),Bytes.toBytes(uid));relaPuts.add(fanPut);}relaPuts.add(attendPut);//对于同一行数据，可以一次添加多个列relaTable.put(relaPuts);（五）HBaseDao封装更新微博收件箱表信息关注用户TablecontTable=connection.getTable(TableName.valueOf(Constants.CONTENT_TABLE));PutinboxPut=newPut(Bytes.toBytes(uid));for(Stringattend:attends){Scanscan=newScan(Bytes.toBytes(attend+"_"),Bytes.toBytes(attend+"|"));ResultScannerresultScanner=contTable.getScanner(scan);（五）HBaseDao封装定义一个时间戳，以解决多条数据同一时间戳的问题关注用户longts=System.currentTimeMillis();for(Resultresult:resultScanner){inboxPut.addColumn(Bytes.toBytes(Constants.INBOX_TABLE_CF),Bytes.toBytes(attend),ts++,result.getRow());}}（五）HBaseDao封装判断uid所关注的人是否有发过微博关注用户if(!inboxPut.isEmpty()){TableinboxTable=connection.getTable(TableName.valueOf(Constants.INBOX_TABLE));inboxTable.put(inboxPut);inboxTable.close();}（五）HBaseDao封装取关用户取关用户publicstaticvoiddeleteAttends(Stringuid,String...dels)throwsIOException{if(dels.length<=0){System.out.println("请传入要取关的用户");return;}Connectionconnection=ConnectionFactory.createConnection(Constants.CONFIGURATION);（五）HBaseDao封装更新用户关系表取关用户TablerelaTable=connection.getTable(TableName.valueOf(Constants.RELATION_TABLE));DeleteuidDelete=newDelete(Bytes.toBytes(uid));ArrayList<Delete>attendDeletes=newArrayList<>();for(Stringdel:dels){uidDelete.addColumns(Bytes.toBytes(Constants.RELATION_TABLE_CF1),Bytes.toBytes(del));（五）HBaseDao封装更新用户关系表取关用户DeleteattendDelete=newDelete(Bytes.toBytes(del));attendDelete.addColumns(Bytes.toBytes(Constants.RELATION_TABLE_CF2),Bytes.toBytes(uid));attendDeletes.add(attendDelete);}relaTable.delete(uidDelete);relaTable.delete(attendDeletes);（五）HBaseDao封装更新微博收件箱表取关用户TableinboxTable=connection.getTable(TableName.valueOf(Constants.INBOX_TABLE));DeleteinboxDelete=newDelete(Bytes.toBytes(uid));for(Stringdel:dels){inboxDelete.addColumns(Bytes.toBytes(Constants.INBOX_TABLE_CF),Bytes.toBytes(del));}inboxTable.delete(inboxDelete);（五）HBaseDao封装创建Get对象，传入uid作为行键，并设置最大版本。传入收件箱表对象，返回一个Result对象，其中包含uid所关注的所有用户的若干条微博的行键。获取某个人的初始化页面TablecontTable=connection.getTable(TableName.valueOf(Constants.CONTENT_TABLE));TableinboxTable=connection.getTable(TableName.valueOf(Constants.INBOX_TABLE));GetinboxGet=newGet(Bytes.toBytes(uid));inboxGet.setMaxVersions();//为Get对象设置最大版本Resultresult=inboxTable.get(inboxGet);（五）HBaseDao封装获取result对象的所有Cell对象，每一个Cell对象包含一条微博数据的一个行键。遍历所有Cell对象，获取微博数据的行键，之后再微博内容表中根据行键找到相应微博数据，此时返回一个Result对象。这个Result对象只包含一个Cell对象（Cell是HBase中的概念，是由列族、列名、时间戳唯一确定的单元），获取这个Cell对象，进一步地获取行键、列族、列名、值的信息。这里的难点在于对Result和Cell的理解。获取某个人的初始化页面（五）HBaseDao封装获取某个人的初始化页面数据获取某个人的初始化页面for(Cellcell:result.rawCells()){GetcontGet=newGet(CellUtil.cloneValue(cell));ResultcontResult=contTable.get(contGet);for(CellcontCell:contResult.rawCells()){System.out.println("RK:"+Bytes.toString(CellUtil.cloneRow(contCell))+",ColumnFamily:"+Bytes.toString(CellUtil.cloneFamily(contCell))+",ColumnName:"+Bytes.toString(CellUtil.cloneQualifier(contCell))+",value:"+Bytes.toString(CellUtil.cloneValue(contCell)));}}（五）HBaseDao封装可以使用Scan起始终止行键的方法，也可以使用过滤器的方法。这里创建一个行键过滤器RowFilter，第一个参数传入比较符，第二个参数传入要比较的对象。之后通过setFilter()方法为Scan对象添加过滤器。获取某个人的所有微博Scanscan=newScan();//构建过滤器RowFilterrowFilter=newRowFilter(CompareFilter.CompareOp.EQUAL,newSubstringComparator(uid+"_"));scan.setFilter(