首页 诗词 字典 板报 句子 名言 友答 励志 学校 网站地图
当前位置: 首页 > 教程频道 > 其他教程 > 互联网 >

hbase shell scan下令扩展

2012-09-20 
hbase shell scan命令扩展也许你还不清楚我在说什么,看一下下面的截图你就明白了: 好吧,如果您感兴趣,可以

hbase shell scan命令扩展

也许你还不清楚我在说什么,看一下下面的截图你就明白了:

 

hbase shell scan下令扩展


好吧,如果您感兴趣,可以继续看下去了。


hbase是以字节数组的形式存储数据的,当你直接用API或通过hbase 自带的shell端去查询数据时,实际显示的是二进制数据的byteString的形式,就像这样:\xE5\x94\xAE\xE5\x90\x8E,当然,这肯定不是你想看到的结果。不过没办法,因为rowkey和value是用户自己定义的,包括类型,长度等,又因为存的是二进制数据,所以hbase自身api不可能解析出真实的数据,因为它不知道存取的规则。

我们要做的,就是要制定这种数据存储的规则,比如:rowkey由几个字段构成,每一个的类型,value的类型。值得一提的是几种数值型一旦类型一定,长度就定了,但string类型比较特殊,长度不是固定的。这里先使用一个字节存储长度,再紧跟着具体string串,当然这只是一种方式。value使用一个columfamily,因为官方也不建议使用太多。

以上的规则作为元数据放在xml配置文件里再好不过了,以后增加了hbase表,只需修改配置文件。

根据上面的规则写几个包含处理逻辑的类是必须的,包括读元数据,类型之间的转换方法等等。有了这些方法,就能在任何地方输出想要的数据,比如web前端。下面是以scan为例的hbase shell客户端具体实现思路:


1. 首先要在ruby/shell/commands下增加一个命令,姑且叫做superscan吧,直接对应一个JRuby脚本,定义一个Superscan类继承自Command,接收结果集并格式化输出。

2.shell.rb脚本中的dml命令组里要添加上面定义的命令。

hbase shell scan下令扩展

3.下面主要是对ruby/hbase/table.rb脚本的修改,添加相应的方法:

因为我的api里用到了表的名字,所以要在初始化方法里加个tablename变量:

def initialize(configuration, table_name, formatter)      @table = org.apache.hadoop.hbase.client.HTable.new(configuration, table_name)      @tableName = table_name    end


superscan方法:


#----------------------------------------------------------    # superScans whole table or a range of keys and returns rows matching specific criterias    def superscan(args = {})      unless args.kind_of?(Hash)        raise ArgumentError, "Arguments should be a hash. Failed to parse #{args.inspect}, #{args.class}"      end      limit = args.delete("LIMIT") || -1      maxlength = args.delete("MAXLENGTH") || -1      if args.any?        filter = args["FILTER"]        startrow = args["STARTROW"] || ''        stoprow = args["STOPROW"]        timestamp = args["TIMESTAMP"]        columns = args["COLUMNS"] || args["COLUMN"] || get_all_columns        cache = args["CACHE_BLOCKS"] || true        versions = args["VERSIONS"] || 1        timerange = args[TIMERANGE]        # Normalize column names        columns = [columns] if columns.class == String        unless columns.kind_of?(Array)          raise ArgumentError.new("COLUMNS must be specified as a String or an Array")        end        scan = if stoprow          org.apache.hadoop.hbase.client.Scan.new(startrow.to_java_bytes, stoprow.to_java_bytes)        else          org.apache.hadoop.hbase.client.Scan.new(startrow.to_java_bytes)        end        columns.each do |c|           family, qualifier = parse_column_name(c.to_s)          if qualifier            scan.addColumn(family, qualifier)          else            scan.addFamily(family)          end        end        unless filter.class == String          scan.setFilter(filter)        else          scan.setFilter(org.apache.hadoop.hbase.filter.ParseFilter.new.parseFilterString(filter))        end        scan.setTimeStamp(timestamp) if timestamp        scan.setCacheBlocks(cache)        scan.setMaxVersions(versions) if versions > 1        scan.setTimeRange(timerange[0], timerange[1]) if timerange      else        scan = org.apache.hadoop.hbase.client.Scan.new      end      # Start the scanner      scanner = @table.getScanner(scan)      count = 0      res = {}      iter = scanner.iterator      # Iterate results      while iter.hasNext        if limit > 0 && count >= limit          break        end        row = iter.next        key = org.apache.hadoop.hbase.util.Bytes::toStringBinary(row.getRow)   #\00\x01这种形式byteKey = row.getRow#addrowType = com.cuirong.bi.data.hbase.reader.MetaConfig::getRowKeyType(@tableName)stringKey = com.cuirong.bi.data.hbase.inf.CommonUtil::bytes2String(byteKey,rowType) #endadd        row.list.each do |kv|          family = String.from_java_bytes(kv.getFamily) #字节数组转string          qualifier = org.apache.hadoop.hbase.util.Bytes::toStringBinary(kv.getQualifier)           column = "#{family}:#{qualifier}"          cell = to_strings(column, kv, byteKey, maxlength)          if block_given?            yield(stringKey, "column=#{column}, #{cell}")          else            res[stringKey] ||= {}            res[stringKey][column] = cell          end        end        # One more row processed        count += 1      end      return ((block_given?) ? count : res)    end

上面的方法用到了to_strings方法:


# Make a String of the passed kv    # Intercept cells whose format we know such as the info:regioninfo in .META.    def to_strings(column, kv, byteKey, maxlength = -1)      if is_meta_table?        if column == 'info:regioninfo' or column == 'info:splitA' or column == 'info:splitB'          hri = org.apache.hadoop.hbase.util.Writables.getHRegionInfoOrNull(kv.getValue)          return "timestamp=%d, value=%s" % [kv.getTimestamp, hri.toString]        end        if column == 'info:serverstartcode'          if kv.getValue.length > 0            str_val = org.apache.hadoop.hbase.util.Bytes.toLong(kv.getValue)          else            str_val = org.apache.hadoop.hbase.util.Bytes.toStringBinary(kv.getValue)           end          return "timestamp=%d, value=%s" % [kv.getTimestamp, str_val]        end      end#addreader = com.cuirong.bi.data.hbase.reader.HBaseReader.new(@tableName)row = reader.getRow(byteKey)#endadd      #val = "timestamp=#{kv.getTimestamp}, value=#{org.apache.hadoop.hbase.util.Bytes::toStringBinary(kv.getValue)}"  val = "timestamp=#{kv.getTimestamp}, value=#{row.getColumn(column)}"      (maxlength != -1) ? val[0, maxlength] : val    end


上面代码中有几个java类是自己定义的,跟据名字应该能猜出做了哪些工作。


可能有些人还不清楚hbase,ruby,JRuby,java之间的关系,下面简单介绍下:

hbase当然是用java实现的开源数据库。

我们常说的ruby是一门Matz(此人跟苍老师同一国籍,顺便喊下口号:钓鱼岛是中国的!)设计的C语言实现解释器的语言,ruby on rails,可以用于web开发。

JRuby是一个用纯java实现了ruby语法解释器的语言,表面上在写ruby脚本,实际运行的还是在jvm上运行字节码文件。JRuby API里面有一个org.jruby.Main 类,作为ruby脚本的入口。

在hbase中, bin/hirb.rb就是入口脚本,它会加载其它ruby脚本,被JRuby API编译成符合jvm规范的字节码文件执行。也许很多工作很久的java工程师都不太清楚的一点:java语言和jvm虚拟机是两个不同的规范,概念。

所谓的Jython,groove都跟JRuby是类似的。


热点排行