Hbase 基础API
Posted 麻辣小虾
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Hbase 基础API相关的知识,希望对你有一定的参考价值。
本文参考:https://www.cnblogs.com/skyl/p/4803738.html
package com.Hbase import org.apache.hadoop.hbase._ import org.apache.hadoop.hbase.client._ import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp import org.apache.hadoop.hbase.filter.SingleColumnValueFilter object HbaseDDL { def main(args: Array[String]): Unit = { // 确认表是否存在 def ensureHbaseTableExist(tableName:TableName) = { // 配置 Hbase val hbaseconf = HBaseConfiguration.create() //val zkConn = "bigdata:2181, bigdata:2182, bigdata:2183" //hbaseconf.set("hbase.zookeeper.quorum", zkConn) val HbaseConn = ConnectionFactory.createConnection(hbaseconf) val adminHbase = HbaseConn.getAdmin() val ifExist = adminHbase.tableExists(tableName) ifExist } // 确认表是否存在 测试 // val result = ensureHbaseTableExist("user_info1") // if (result) { // println("表存在") // } else { // println("表不存在") // } /** * Hbase建表 两个参数 * @param tableName 形式为 ns:tb 或者 tb API 创建 namespace 机会不多,一般通过 hbase shell 创建 * @param columnFamilys cf1,cf2,cf3 */ def createHbaseTable(tableName:String, columnFamilys:String) = { // 配置 Hbase val hbaseconf = HBaseConfiguration.create() //val zkConn = "bigdata:2181, bigdata:2182, bigdata:2183" //hbaseconf.set("hbase.zookeeper.quorum", zkConn) val HbaseConn = ConnectionFactory.createConnection(hbaseconf) val adminHbase = HbaseConn.getAdmin() // 列簇逗号分隔 val CFS = columnFamilys.split(",") // 表名,判断是否带了namespace,带了则判断是否存在 namespace, 不存在则创建 val nameSpace = tableName.split(":")(0) if (nameSpace != tableName) { adminHbase.createNamespace(NamespaceDescriptor.create(tableName.split(":")(0)).build()) println("NameSpace 创建成功!") } // 判断表是否存在,不存在新建,存在则提示 if (!ensureHbaseTableExist(TableName.valueOf(tableName))) { // 实例化 HTableDescriptor val htable = new HTableDescriptor(TableName.valueOf(tableName)) // 循环添加所有列簇 for ( columnFamily <- CFS) { // 实例化 HColumnDescriptor val htableColumnFamily1 = new HColumnDescriptor((columnFamily)) // 调用 HColumnDescriptor 设置列簇属性 htableColumnFamily1.setMaxVersions(3) // 表增加列族 htable.addFamily(new HColumnDescriptor(columnFamily)) } // 创建表 adminHbase.createTable(htable) println("表创建成功") } else { println("表已存在") } adminHbase.close() } // 测试建表 // createHbaseTable("scTable3", "info,base") /** * 列出所有表 */ def listAllHbaseTable() ={ // 配置 Hbase val hbaseconf = HBaseConfiguration.create() //val zkConn = "bigdata:2181, bigdata:2182, bigdata:2183" //hbaseconf.set("hbase.zookeeper.quorum", zkConn) val HbaseConn = ConnectionFactory.createConnection(hbaseconf) val adminHbase = HbaseConn.getAdmin() val listTables = adminHbase.listTableNames() for(table <- listTables){ println(table) } adminHbase.close() } //listAllHbaseTable() /** * 删除一张表,输入表名 * 判断是否存在,是否失效,否则不能删除 * * @param tableName */ def deleteHbaseTable(tableName: String) ={ val hbaseconf = HBaseConfiguration.create() val HbaseConn = ConnectionFactory.createConnection(hbaseconf) val adminHbase = HbaseConn.getAdmin() val tbName = TableName.valueOf(tableName) if(ensureHbaseTableExist(tbName)){ // 若表不失效,则使失效 if(!adminHbase.isTableDisabled(tbName)){ adminHbase.disableTable(tbName) } adminHbase.deleteTable(tbName) println("删除成功") } else { println("表不存在") } adminHbase.close() } //deleteHbaseTable("scTable3") /** * * 删除表的某个列族 ,得到 HTableDescriptor , 调用该类的 removeFamily 方法 * @param tableName 表名 ---> ns:tb or tb [String] * @param columnFamily 列族名 ---> String */ def deleteHbaseColumnFamily(tableName:String, columnFamily:String) ={ val hbaseconf = HBaseConfiguration.create() val HbaseConn = ConnectionFactory.createConnection(hbaseconf) val adminHbase = HbaseConn.getAdmin() val tbName = TableName.valueOf(tableName) // disable table adminHbase.disableTable(tbName) // get HTableDescriptor val htd = adminHbase.getTableDescriptor(tbName) // delete family htd.removeFamily(columnFamily.getBytes()) // apply htd to table adminHbase.modifyTable(tbName, htd) // enable table adminHbase.enableTable(tbName) println("删除成功") adminHbase.close() } // deleteHbaseColumnFamily("scTable3", "base") /** * 给表增加列族 先得到表的 HTableDescriptor, 然后使用 HColumnDescriptor 初始化 新增列,并设置属性 * 调用 HTableDescriptor 的 addFamily 方法,将初始化好的 HCD 添加到 HTableDescriptor ,然后使用admin 的 modifyTable 方法将修改应用 * @param tableName * @param columnFamily */ def addHbaseColumnFamily(tableName:String, columnFamily:String) ={ val hbaseconf = HBaseConfiguration.create() val HbaseConn = ConnectionFactory.createConnection(hbaseconf) val adminHbase = HbaseConn.getAdmin() val tbName = TableName.valueOf(tableName) // disable table adminHbase.disableTable(tbName) // get HTableDescriptor val htd = adminHbase.getTableDescriptor(tbName) // val hcd = new HColumnDescriptor(columnFamily) hcd.setMaxVersions(3) // add family htd.addFamily(hcd) // apply htd to table adminHbase.modifyTable(tbName, htd) // enable table adminHbase.enableTable(tbName) println("添加成功") adminHbase.close() } //addHbaseColumnFamily("scTable3", "base") /** * 修改列簇功能 get 到 HTableDescriptor ,再 get 到 Family ,设置 Family ,admin modifyTable 应用 * @param tableName * @param columnFamily */ def modifyHbaseTableColumnFamily(tableName:String, columnFamily:String) ={ val hbaseconf = HBaseConfiguration.create() val HbaseConn = ConnectionFactory.createConnection(hbaseconf) val adminHbase = HbaseConn.getAdmin() val tbName = TableName.valueOf(tableName) adminHbase.disableTable(tbName) val htd = adminHbase.getTableDescriptor(tbName) val modifyCol = htd.getFamily(columnFamily.getBytes()) modifyCol.setMaxVersions(3) adminHbase.modifyTable(tbName,htd) adminHbase.enableTable(tbName) adminHbase.close() println("修改成功!") } //modifyHbaseTableColumnFamily("scTable3", "info") /** * 插入数据,五个参数 * @param tableName * @param columnFamily * @param column * @param rowkey * @param value */ def putDataHbaseTable(tableName:String, columnFamily:String, column:String, rowkey:String, value:String) ={ val hbaseconf = HBaseConfiguration.create() // table val hTable = new HTable(hbaseconf, tableName) // row key val putData = new Put(rowkey.getBytes()) // put value putData.add(columnFamily.getBytes(), column.getBytes(), value.getBytes()) /** * 插入方式 * ASYNC_WAL : 当数据变动时,异步写WAL日志 * SYNC_WAL : 当数据变动时,同步写WAL日志 * FSYNC_WAL : 当数据变动时,同步写WAL日志,并且,强制将数据写入磁盘 * SKIP_WAL : 不写WAL日志 * USE_DEFAULT : 使用HBase全局默认的WAL写入级别,即 SYNC_WAL */ putData.setDurability(Durability.SYNC_WAL) // put data to table hTable.put(putData) println("插入数据成功!") // close table hTable.close() } //putDataHbaseTable("scTable3", "info", "name", "rk0002", "lalala") def deleteDataHbaseTable(tableName: String, rowkey:String, columnFamily:String, column:String = null)={ val hbaseConf = HBaseConfiguration.create() val hTable = new HTable(hbaseConf, tableName) // 初始化 Delete ,表名后可接时间戳 val deletaData = new Delete(rowkey.getBytes()) /** * 1).删除指定列的 最新版本 的数据:Delete addColumn (byte[] family, byte[] qualifier) * 2).删除指定列的 指定版本 的数据:Delete addColumn (byte[] family, byte[] qualifier, long timestamp ) * 3).删除指定列的 所有版本 的数据:Delete addColumns (byte[] family, byte[] qualifier) * 4).删除指定列的,时间戳 小于等于 给定时间戳的 所有版本 的数据:Delete addColumns (byte[] family, byte[] qualifier, long timestamp ) * 5).删除指定列族的所有列的 所有版本 数据:Delete addFamily (byte[] family) 默认使用当前时间的时间戳,时间戳大于当前时间的数据删除不掉 * 6).删除指定列族的所有列中 时间戳 小于等于 指定时间戳 的所有数据:Delete addFamily (byte[] family, long timestamp) * 7).删除指定列族中 所有列的时间戳 等于 指定时间戳的版本数据:Delete addFamilyVersion (byte[] family, long timestamp) */ deletaData.addColumn(columnFamily.getBytes(),column.getBytes()) //deletaData.addColumns(columnFamily.getBytes(),column.getBytes()) //deletaData.addFamily(columnFamily.getBytes()) hTable.delete(deletaData) println("删除成功") hTable.close() } // deleteDataHbaseTable("scTable3", "rk0002", "info") def getDataHbaseTable(tableName:String, rowkey:String, columnFamily:String, column:String = null)={ val hbaseCOnf = HBaseConfiguration.create() val hTable = new HTable(hbaseCOnf, tableName) val getData = new Get(rowkey.getBytes()) /** * 1). Get addFamily(byte[] family) 指定希望获取的列族 * 2). Get addColumn(byte[] family, byte[] qualifier) 指定希望获取的列 * 3). Get setTimeRange(long minStamp, long maxStamp) 设置获取数据的 时间戳范围 * 4). Get setTimeStamp(long timestamp) 设置获取数据的时间戳 * 5). Get setMaxVersions(int maxVersions) 设定获取数据的版本数 * 6). Get setMaxVersions() 设定获取数据的所有版本 * 7). Get setFilter(Filter filter) 为Get对象添加过滤器 * 8). void setCacheBlocks(boolean cacheBlocks) 设置该Get获取的数据是否缓存在内存中 */ //getData.addFamily(columnFamily.getBytes()) //getData.addColumn(columnFamily.getBytes(), column.getBytes()) //getData.setTimeStamp("1535680422860".toLong) getData.setMaxVersions() val results = hTable.get(getData) for (result <- results.rawCells()){ println( new String(CellUtil.cloneRow(result)) + "\\t" + new String(CellUtil.cloneFamily(result)) + "\\t" + new String(CellUtil.cloneQualifier(result)) + "\\t" + new String(CellUtil.cloneValue(result)) + "\\t" + result.getTimestamp) } hTable.close() } //getDataHbaseTable("scTable", "rk0002", "info", "age") def scanDataHbaseTable(tableName:String, startRow:String, stopRow:String, columnFamily:String, column:String)={ val hBaseConf = HBaseConfiguration.create() val hTable = new HTable(hBaseConf, tableName) /** * 1). 创建扫描所有行的Scan:Scan() * 2). 创建Scan,从指定行开始扫描:Scan(byte[] startRow) * 注意:如果指定行不存在,从下一个最近的行开始 * 3). 创建Scan,指定起止行:Scan(byte[] startRow, byte[] stopRow) * 注意: startRow <= 结果集 < stopRow * 4). 创建Scan,指定起始行和过滤器:Scan(byte[] startRow, Filter filter) */ val scanData = new Scan() val filter1 = new SingleColumnValueFilter(columnFamily.getBytes(), column.getBytes(), CompareOp.GREATER_OR_EQUAL, "60".getBytes() ) /** * Scan setStartRow (byte[] startRow) 设置Scan的开始行,默认 结果集 包含该行。如果希望结果集不包含该行,可以在行键末尾加上0。 * Scan setStopRow (byte[] stopRow) 设置Scan的结束行,默认 结果集 不包含该行。如果希望结果集包含该行,可以在行键末尾加上0。 * Scan setBatch(int batch) 指定最多返回的Cell数目.用于防止一行中有过多的数据,导致OutofMemory错误 * Scan setTimeRange (long minStamp, long maxStamp) 扫描指定 时间范围 的数据 * Scan setTimeStamp (long timestamp) 扫描 指定时间 的数据 * Scan addColumn (byte[] family, byte[] qualifier) 指定扫描的列 * Scan addFamily (byte[] family) 指定扫描的列族 * Scan setFilter (Filter filter) 为Scan设置过滤器,详见HBase API Filter过滤器 * Scan setReversed (boolean reversed) 设置Scan的扫描顺序,默认是正向扫描(false),可以设置为逆向扫描(true)。注意:该方法0.98版本以后才可用!! * Scan setMaxVersions () 获取所有版本的数据 * Scan setMaxVersions (int maxVersions) 设置获取的最大版本数! 不调用上下两个setMaxVersions() 方法,只返回最新版本数据 * void setCaching (int caching) 设定缓存在内存中的行数,缓存得越多,以后查询结果越快,同时也消耗更多内存 * void setRaw (boolean raw) 激活或者禁用raw模式。如果raw模式被激活,Scan将返回 所有已经被打上删除标记但尚未被真正删除 的数据。该功能仅用于激活了 KEEP_DELETED_ROWS的列族,即列族开启了 hcd.setKeepDeletedCells(true) * Scan激活raw模式后,只能浏览所有的列,而不能指定任意的列,否则会报错 */ scanData.setFilter(filter1) val resultsScan:ResultScanner = hTable.getScanner(scanData) while (resultsScan.iterator().hasNext){ val results = resultsScan.iterator().next() for (result:Cell <- results.rawCells()) { println(new String(CellUtil.cloneRow(result)) + "\\t" + new String(CellUtil.cloneFamily(result)) + "\\t" + new String(CellUtil.cloneQualifier(result)) + "\\t" + new String(CellUtil.cloneValue(result)) + "\\t" + result.getTimestamp) } } /** * for 循环无法直接遍历 ResultScanner 暂无办法 */ // for(results:Result <- resultsScan){ // // for (result:Cell <- results.rawCells()) { // // println(new String(CellUtil.cloneRow(result)) + "\\t" + // new String(CellUtil.cloneFamily(result)) + "\\t" + // new String(CellUtil.cloneQualifier(result)) + "\\t" + // new String(CellUtil.cloneValue(result)) + "\\t" + // result.getTimestamp) // } // // } hTable.close() } //scanDataHbaseTable("scTable", "rk0001", "rk0002", "info", "age") } }
以上是关于Hbase 基础API的主要内容,如果未能解决你的问题,请参考以下文章