尚大数据技术之hbase HBase

HBase的原型是的BigTable 思想的启发，目前作为Hadoop的： 1)处理regionHBaseHBase读写数据的时候，数据不是直接写进磁盘，它会在内存中Write-Aheadlogfile的文件中，Hbase表的分片，HBase表会根据RowKey值被切分成不同的region在RegionServer中，在一个RegionServer中可以有多个不同的region。常开发人员比较熟练的是结构化的数据进行处理，但是在HDFS直接的文件往往不具HBaseHDFS上的操作。如果需要查询数据，只需要通过键值HBase内置有Zookeeper，但一般我们会有其他的Zookeeper集群来master和HRegionServer启动时会向ZooKeeper，所有HRegion的寻址，实时HRegionserver的上线和下线信息。并实时通知给HMaster，HBase的schema和table元数据，默认情况下，HBaseZooKeeper实例，ZookeeperHMaster不再是了ActiveHMaster的负担。 、HBase的配置文件需要修改HBase对应的配置文件。hbase-env.sh修改内容：exportexportexport<value>/home/admin/modules/zookeeper-、HBaseJar -- - $bin/hbase-daemon.sh$bin/hbase-daemon.shstart尖叫提示：如果集群之间的节点时间不同步，会导致regionserver启动方式尖叫提示：如果使用的是JDK8hbase-evn.sh中移除Hbse$bin/hbase$bin/hbasehbase(main)>put'student','1001','info: hbase(main)>put'student','1001','info:age','18'hbase(main)>put'student','1002','info:name','Janna'hbase(main)>put'student','1002','info: hbase(main)>puthbase(main)hbase(main)>scanhbase(main)>scan'student',{STARTROW=>'1001',STOPROW=>'1001'}hbase(main)>scan'student',{STARTROW=>'1001'}hbase(main)hbase(main)>puthbase(main)>puthbase(main)>hbase(main)>gethbase(main)>gethbase(main)>delete hbase(main)hbase(main)>delete hbase(main)>deleteallhbase(main)hbase(main)>truncate首先需要先让该表为disablehbase(main)hbase(main)>disable然后才能drophbase(main)hbase(main)>dropdrop表，会报错：DropthenamedtableTablemustfirstbedisabledERROR:Tablestudentisenabled.Disableitfirst.hbase(main)hbase(main)>counthbase(main)hbase(main)>alterHRegionServer保存着meta表以及表数据，要表数据，首先 先去zookeeper，从zookeeper里面获取meta表所在的位置信息，即找到这个meta表在哪个HRegionServer上保存着。通过元数据中的信息，对应的HRegionServer，然后扫描所在HRegionServer的MemstoreStorefileRegionServer服务器发起写入数据请求，然后RegionServer收到请求并响然后将数据写入到MemstoreStorefile越来越多，会触发CompactStorefile合并成一个大的将Region一分为二。exportMaven式。因为在企业中，有些时候我们需要一些过时的API来提供更好的兼容性。conf=Port",}publicpublicstaticbooleanisTableExist(StringtableName)throwsMasterNotRunningException,ZooKeeperConnectionException,IOException{ //HBaseAdminadmin=(HBaseAdmin)connection.getAdmin();HBaseAdminadmin=newHBaseAdmin(conf);}publicpublicstaticvoidcreateTable(StringtableName,String...columnFamily)throwsMasterNotRunningException,ZooKeeperConnectionException,IOException{HBaseAdminadmin=newfor(Stringcf:columnFamily){}}}publicpublicstaticvoiddropTable(StringtableName)throwsMasterNotRunningException,ZooKeeperConnectionException,IOException{HBaseAdminadmin=newHBaseAdmin(conf);}}publicpublicstaticvoidaddRowData(StringtableName,StringrowKey,StringcolumnFamily,Stringcolumn,Stringvalue)throwsIOException{Putput=new}publicpublicstaticvoiddeleteMultiRow(StringtableName,String...rows)throwsIOException{HTablehTable=newHTable(conf,tableName);List<Delete>dele ist=newArrayList<Delete>();for(Stringrow:Deletedelete=newDelete(Bytes.toBytes(row)); } }publicstaticvoidgetAllRows(StringtableName)throwsIOException{publicstaticvoidgetAllRows(StringtableName)throwsIOException{HTablehTable=newHTable(conf,tableName);Scanscan=new//使用HTable得到resultcanner实现类的对象for(Resultresult:resultScanner){Cell[]cells=result.rawCells();for(Cellcell:cells){System.out.println("列族Bytes.toString(CellUtil.cloneFamily(cell)));System.out.println("列Bytes.toString(CellUtil.cloneQualifier(cell)));System.out.println("值:"+Bytes.toString(CellUtil.cloneValue(cell)));}}}publicpublicstaticvoidgetRow(StringtableName,StringrowKey)throwsIOException{HTabletable=newHTable(conf,tableName);Getget=newGet(Bytes.toBytes(rowKey));Resultresult=table.get(get);for(Cellcell:System.out.println("行键:"+Bytes.toString(result.getRow()));System.out.println("列族Bytes.toString(CellUtil.cloneFamily(cell)));System.out.println("值:"+Bytes.toString(CellUtil.cloneValue(cell)));System.out.println("时间戳:"+cell.getTimestamp());}}publicpublicstaticvoidgetRowQualifier(StringtableName,StringrowKey,Stringfamily,Stringqualifier)throwsIOException{Getget=newResultresult=table.get(get);for(Cellcell:System.out.println("行键:"+Bytes.toString(result.getRow()));System.out.println("列族Bytes.toString(CellUtil.cloneFamily(cell)));System.out.println("值:"+Bytes.toString(CellUtil.cloneValue(cell)));}}HBaseJavaAPIHBaseMapReduce过程，比如使用MapReduce将数据从本地文件系统导入到HBase的表中，比如我们从HBase中一些原始数据后使用MapReduce做数据分析。 $bin/hbaseHBase$bin/hbase运行的MapReduce任--案例二：使用MapReducetsvHDFS中创建input_fruitfruit.tsvHBase- packagepackageimportorg.apache.hadoop.hbase.Cell;importorg.apache.hadoop.hbase. .Put;importorg.apache.hadoop.hbase. importorg.apache.hadoop.hbase.mapreduce.TableMapper;importorg.apache.hadoop.hbase.util.Bytes;throwsIOException,InterruptedException{Putput=newfor(Cellcell:}else}}}}}packageimportorg.apache.hadoop.hbase.mapreduce.TableReducer;importNullWritable>{protectedvoidreduce(ImmutableBytesWritablekey,I ble<Put>packageimportorg.apache.hadoop.hbase.mapreduce.TableReducer;importNullWritable>{protectedvoidreduce(ImmutableBytesWritablekey,I ble<Put>values,Contextcontext)throwsIOException,InterruptedException{for(Put}}Configurationconf=Scanscan=newScan();"fruit",//数据源的表名ReadFruitMapper.class,//设置Mapper类Put.class,//Mappervalue值类型

booleanisSuccess thrownewIOException("Jobrunningwith}returnisSuccess?0:}publicstaticvoidpublicstaticvoidmain(String[]args)throwsException{Configurationconf=尖叫提示：maven打包命令：-Plocalcleanpackage或-Pdevcleanpackageinstall（将第jar包一同打包，需要插件：maven-shade-plugin）HBase-目标：实现将HDFS中的数据写入到HBasepackagepackageimportorg.apache.hadoop.hbase.util.Bytes;importorg.apache.hadoop.io.Text;ImmutableBytesWritable,Put>{protectedvoidmap(LongWritablekey,Textvalue,Contextcontext)throwsIOException,InterruptedException{String[]values=StringrowKey=values[0];Stringname=values[1];Stringcolor=values[2];Putput=newput.add(Bytes.toBytes("info"),put.add(Bytes.toBytes("info"),Bytes.toBytes("name"),Bytes.toBytes(name)); }}packagepackageimportorg.apache.hadoop.hbase.mapreduce.TableReducer;importNullWritable>{protectedvoidreduce(ImmutableBytesWritablekey,I ble<Put>values,Contextcontext)throwsIOException,InterruptedException{for(Put}}Txt2FruitRunnerConfigurationconf=FileInputFormat.addInputPath(job,inPath);//设置MapperbooleanisSuccess= }returnisSuccess?0:}publicstaticpublicstaticvoidmain(String[]args)throws{Configurationconf=尖叫提示：maven打包命令：-Plocalcleanpackage或-Pdevcleanpackageinstall（将第jar包一同打包，需要插件：maven-shade-plugin）Hive、HBaseHive便使用HQL去管理查询。用 数据持久化的体现形式是Hfile，存放于DataNode中，被ResionServer以region的形式、HBaseHive尖叫提示：HBase与Hive的集成在的两个版本中无法兼容。所以，我们只能含着泪勇HBase的JarHive所依赖的Jar包（或者使用软连接的形式）$ln-s$ln-s-1.3.1.jar-$ln-s$ln-s<description>Thelistof<description>ThelistofZooKeeperserverstotalkto.Thisisonlyneededforread/write<description>TheportofZooKeeperserverstotalkto.Thisisonlyneededforread/writeHiveenamestring,jobstring,mgrint,saldouble,commdouble,deptnoWITHSERDEPROPERTIES "empnoint,enamestring,jobstring,mgrint,saldouble,commdouble,deptnoint)rowformatdelimitedfieldsterminatedbyHiveloadhive>insertintotablehive_hbase_emp_tableselecthive>insertintotablehive_hbase_emp_tableselect*fromhive>selecthive>select*fromhbase>hbase>scanenamestring,jobstring,mgrint,saldouble,commdouble,deptnoint)STOREDBYWITHSERDEPROPERTIES("hbase.columns.map hivehive(default)>select*fromSqoopSqoopsupportsadditionalimporttargetsbeyondHDFSandHive.SqoopcanalsoimportrecordsintoatableinHBase.之前我们已经学习过如何使用Sqoop在Hadoop集群和关系型数据库中进行数据的导入导出工作，接下来我们学下利用Sqoop在HBase和RDBMS中进行数据的转储。SetsthetargetcolumnfamilyfortheIfspecified,createmissingHBaseSpecifieswhichinputcolumntouseastherowkey.Incase,ifinputtablecontainscompositekey,then<col>mustbeintheformofacomma-separatedlistofcompositekeymysqlHBase避免rowkey的重复）SpecifiesanHBasetabletouseasthetargetinsteadofHDFS.指定数据将要导入到HBase中的哪中Enablesbulk1)目标：RDBMS中的数据抽取到HBase配置在Mysql中新建一个数据库db_library， nameVARCHAR(255)NOTNULL,INSERTINTOdb_library.book(name,price)VALUES('LieSporting','30');INSERTINSERTINTOdb_library.book(name,price)VALUES('LieSporting','30');INSERTINTOdb_library.book(name,price)VALUES('Pride&Prejudice','70');$$bin/sqoopimport--usernameroot--password123456--tablebook--num-mappers1hbase>create尖叫提示：sqoop1.4.6HBase1.0.1之前的版本的自动创建HBasehbase>create在HBase中scan这得到如下内hbase>hbase>scan、常用的 hbase>hbase>statushbase>hbase>hbase>hbase>hbase>hbase>counthbase>hbase>describehbase>hbase>existhbase>hbase>alter'hbase_book',NAME=>'CF2',VERSIONS=>hbase>hbase>alter'hbase_book','delete'=>hbase>hbase>disablehbase>hbase>drophbase>hbase>truncatehbase>hbase>create‘table’,hbase>hbase>create't1',{NAME=>'f1'},{NAME=>'f2'},{NAME=> $$bin/hadoopdistcphdfs://linux01:8020/hbase\$$bin/hadoopdistcp\当启动regionserver时，regionserver会向HMaster并开始接收本地数据，开始的时候，新加入的节点不会有任何数据，平衡器开启的情况下，将会有新的region移动到开启的RegionServer上。如果启动和停止进程是使用ssh和HBase，那么会将新添加的节点的主机名加入到conf/regionservers文件中。 在退役节点上停止hbase>hbase>hbase-daemon.shstopRegionServer一旦停止，会关闭的所有Master节点检测到该RegionServer另案2~6cell。入数据时，的值会将最老的值覆盖。（现版本已默认为1）在HBase中Hmaster负责RegionServer的生命周期，均衡RegionServer的负载，如果持太久。所以HBase支持对Hmaster的高可用配置。HBase集群（如果没有开启则跳过此步$touchconf/backup-在 $touchconf/backup-$$echolinux02>conf/backup-将整个 NameNode元数据备份使用 NameNodedir整为50~1005~10即可。调整为3~5。true可以解释：以上两个属性分别为一个单独的Job解释：以上两个属性分别为一个单独的JobMapReduce在同一时刻，不要同时运行太多的MapReduce，这样会消耗过多的内存，任务会执行的非IO，提高性map=2+⅔cpu_corereduce=2+⅓cpu_core$sudoblockdev$sudoblockdev--setra32768$$sudosysctl-w$ulimitn$ulimitn$ulimitu**--*开启集群的时间同步解释：Inhbase-site.xml,setzookeeper.session.timeoutto30secondsorlesstoboundfailuredetection(20-30secondsisagoodstart).该值会直接关系到master发现服务器宕机的最大周期，默认值为30秒，如果该值过小，会在HBase在写入大量数据发生而GC时，导致每一个region着startRow与endRowKey，如果加入的数据符合某个region的rowKey范围，则该数据交给这个region。那么依照这个原则，我们可以将数据索要投放的分区提前大致的规划好，以提高HBase性能。splits.txt=HBaseAdminhAdmin=newrowkeyregion中，在一定程度上防止数据倾斜。接下来我们就谈一谈rowkey常用的设计方案。rowKey1001的，SHA1rowKey3001的，SHA1 置为4096或者更高。默认值：4096值设置为更大的值（默认60000毫秒），以确保socket不会被timeout掉。DataNodeDataNode设置RPC数量解释：默认值为30，用于指定RPC HFileregionHfile。hbase scan.nextHBase Storefile文件。splitRegionRegion即：这个参数的作用是当单个HRegion内所有的Memstore大小总和超过指定值时，flushHRegion的所有memstore。RegionServer的flush是通过将请求添加一个队列，模拟时，可能会导致内存陡增，的情况是触发OOM。值时，将会有多个MemStoresflush到文件中，MemStoreflush顺序是按照大小降序执行的，直到刷新到MemStore使用内存略小于lowerLimit则在default默认名空间中。观察HDFS中 、系privateConfigurationconf=@parampublicinitNamespace(){HBaseAdminadmin=null;try{Connectionconnection=ConnectionFactory.createConnection(conf);admin=(HBaseAdmin)connection.getAdmin();NamespaceDescriptorweibo=NamespaceDescriptor{{admin){try{}}}}1**createTableContent(){HBaseAdminadmin=null;Connectionconnection=null;try{admin=(HBaseAdmin)connection.getAdmin();HTableDescriptorcontentTableDescriptor=newHColumnDescriptorinfoColumnDescriptor=new {}admin){try{}}}}1HBaseAdminadmin=null;try{Connectionconnection=ConnectionFactory.createConnection(conf);admin=(HBaseAdmin)connection.getAdmin();HTableDescriptorrelationTableDescriptor=newHColumnDescriptorattendColumnDescriptor=new {{{admin){try{}}}}** *createTableInbox(){HBaseAdminadmin=null;try{Connectionconnection=admin=(HBaseAdmin)HTableDescriptorinboxTableDescriptor=new{{{admin){try{}}}}packagepublicclass{privateStringuid;privateStringcontent;publicStringgetUid()return}{this.uid=}return}{this.timestamp=}}{this.content=}return"Message[uid="+uid+",timestamp="+timestamp+",content="+content}} *b *bpublicvoidpublishContent(Stringuid,Stringcontent){Connectionconnection=null;try{connection= longtimestamp=System.currentTimeMillis();StringrowKey=uid+"_"+timestamp;Putput=newPut(Bytes.toBytes(rowKey));Getget=newGet(Bytes.toBytes(uid));Resultresult=relationTable.get(get);for(Cellcell:}if(fans.size()<=0)return;List<Put>puts=newArrayList<Put>();for(byte[]fan:fans){PutfansPut=new}}catch(IOExceptione)connection){try{{}}}}**a***a*b*cif(attends==null||attends.length<=0||uid==null||uid.length()0){}Connectionconnection=null;try{connection=List<Put>puts=newArrayList<Put>();PutattendPut=newPut(Bytes.toBytes(uid));for(Stringattend:attends){PutfansPut=newPut(Bytes.toBytes(attend));}Scanscan=newScan();List<byte[]>rowkeys=newArrayList<byte[]>();for(Stringattend:RowFilterfilter=newRowFilter( parator(attend+"_"));ResultScannerresult= tor<Result>i tor=result.i Resultr=i for(Cellcell:}}}if(rowkeys.size()<=0)for(byte[]rk:rowkeys){Putput=newStringrowKey=put.addColumn(Bytes.toBytes("info"),Bytes.toBytes(attendUID),timestamp,rk);}}catch(IOExceptione){connection){try{}catch(IOExceptione)//TODOAut