基于HBase的冠字号查询系统1--理论部分
1. 软件版本和部署
maven：3.3.9，jdk：1.7 ，Struts2：2.3.24.1，hibernate：4.3.6，spring：4.2.5，MySQL：5.1.34，Junit：4，Myeclipse：2014;
Hadoop2.6.4，HBase1.1.2
下载：https://github.com/fansy1990/ssh_v3/releases
数据下载：http://download.csdn.net/detail/fansy 或
2. 背景&思路
目前针对钞票识别，一般都是使用看、摸、听、测四种方式，这里使用一种比较客观的方式类进行识别。 建设冠字号管理查询，以冠字号查询为手段，有效解决银行对外误付假币的问题。从源头解决伪钞问题。
本系统就是使用客观的方法来验证伪钞。本系统采用的方案是基于冠字号的，每张人民币的冠字号是唯一的，如果有一个大表可以把所有的人民币以及人民币对应的操作（在什么时间、什么地点存入或获取）记录下来，这样在进行存取时就可以根据冠字号先查询一下，看当前冠字号对应的纸币在大表中的保存的情况，这样就可以确定当前冠字号对应的纸币是否是伪钞了（这里假设在大表中的所有冠字号对应的钞票都是真钞）。
下面对应存储场景：
最近状态（表中有无）
有（此时没有无状态）
目前，基于传统存储数据一般在千万级别（受限于查询等性能），但是如果要存储所有钞票的信息以及其被存储或获取的记录信息，那么传统数据库肯定是不能胜任的。所以本系统是基于HBase的。
3. 功能指标
? 存储万级用户信息；
? 存储百万级别钞票信息；
? 支持前端业务每秒500+实时查询请求；
? 数据存储和计算能够可扩展；
? 提供统一接口，支持前端相关查询业务；
说明： 其中前两条，万级用户信息和百万级钞票信息是根据数据确定的，这里可以根据数据以及集群的大小进行调整（如果集群够大，存储信息也可以很大）；
冠字号查询系统包括下面5层：
? 数据层：包括基础数据MySQL、文档、Web数据等；
? 数据处理层：主要是数据的加载，包括MR加载方式、 API加载模式、Sqoop加载模式等；
? 数据存储层：主要是HBase存储，包括钞票的所有信息以及用户信息等；
? 数据服务层：主要是对外提供查询、存储等接口服务；
? 数据应用层：存取钞系统，在存钞时设计到伪钞识别；其他应用系统；
5.1原始数据：
冠字号存储记录（冠字号，表中是否有该冠字号（0表示没有，1表示有），存储或取时间，存储或取所在银行编号，用户id）：
用户信息表（用户Id，名字，出生日期，性别，地址，手机号，绑定银行编号）：
5.2冠字号记录
对数据进过初步探索，发现冠字号规律如下：
AAA[A~Z][]
AAB[A~Z][]
如果集群有四个节点，设置region初始为4，那么三个split点为：AAAM9999，AAAZ9999，AABM9999；
假设每个用户每天进行10次操作，如果要保存100天数据，那么设置版本数为1000，则建表语句如下：
create 'records',{NAME=&'info',VERSIONS=&1000},SPLITS =&['AAAM9999','AAAZ9999','AABM9999']
表结构描述如下：
字段值举例
表主键（钞票冠字号）
long型（可以存储用户操作的时间）
who、when、where做了哪些操作
如果用户是存储行为，那么在行为结束后，该值为1
存取钞银行
5.3用户信息
对数据进过初步探索，发现用户信息规律如下：
~92]XXXX[]
如果集群有四个节点，设置region初始为4，那么三个split点为：XXXX1991XXXX1992XXXX0000；
则建表语句如下：
create 'user',{NAME=&'info'},SPLITS =&['XXXX0000','XXXX0000','XXXX0000']
表结构描述如下:
字段值举例
用户主键（身份证号）
用户注册银行
用户出生年月
6. 数据加载
系统在投入使用的时候，已经存在历史数据，需要把历史数据批量导入到系统中；在人民币首次发行时，也需要批量导入系统中。这里的导入直接使用MR导入。
MR设计成一个通用的数据从HDFS导入HBase的MR：
6.1 主类：
package ssh.
import org.apache.hadoop.conf.C
import org.apache.hadoop.conf.C
import org.apache.hadoop.fs.P
import org.apache.hadoop.hbase.TableN
import org.apache.hadoop.hbase.mapreduce.TableMapReduceU
import org.apache.hadoop.mapreduce.J
import org.apache.hadoop.mapreduce.lib.input.FileInputF
import org.apache.hadoop.mapreduce.lib.input.TextInputF
import org.apache.hadoop.util.T
import ssh.util.HadoopU
* Job Driver驱动类
* @author fansy
public class ImportToHBase extends Configured implements Tool {
public static final String SPLITTER = &SPLITTER&;
public static final String COLSFAMILY = &COLSFAMILY&;
public static final String DATEFORMAT = &DATEFORMAT&;
public int run(String[] args) throws Exception {
if (args.length != 5) {
System.err
.println(&Usage:\n demo.job.ImportToHBase
return -1;
if (args[3] == null || args[3].length() & 1) {
System.err.println(&column family can't be null!&);
return -1;
Configuration conf = getConf();
conf.set(SPLITTER, args[2]);
conf.set(COLSFAMILY, args[3]);
conf.set(DATEFORMAT, args[4]);
TableName tableName = TableName.valueOf(args[1]);
Path inputDir = new Path(args[0]);
String jobName = &Import to & + tableName.getNameAsString();
Job job = Job.getInstance(conf, jobName);
job.setJarByClass(ImportMapper.class);
FileInputFormat.setInputPaths(job, inputDir);
job.setInputFormatClass(TextInputFormat.class);
job.setMapperClass(ImportMapper.class);
TableMapReduceUtil.initTableReducerJob(tableName.getNameAsString(),
null, job);
job.setNumReduceTasks(0);
HadoopUtils.setCurrJob(job);// 设置外部静态Job
return job.waitForCompletion(true) ? 0 : 1;
主类的run方法中使用的是传统的MR导入HBase的代码，只是设置了额外的参数，这里主类参数意思解释如下：
input: HDFS输入数据路径;
splitter : 输入数据字段分隔符;
tableName : 表名;
: 列描述， rk代表rowkey以及rowkey所在列、ts代表timestamp及其所在列；示例数据说明原始数据，第一列为rowkey，第二列为timestamp所在列，第三列属于列簇col1，同时列名为q1，第4列属于列簇col2,同时列名为q1；
date_format : timestamp日期格式，如果列描述中没有ts那么就代表原始数据中没有timestamp，则此参数没有意义；
6.2 Mapper：
package ssh.
import java.io.IOE
import java.text.ParseE
import java.text.SimpleDateF
import java.util.ArrayL
import org.apache.hadoop.hbase.client.P
import org.apache.hadoop.hbase.io.ImmutableBytesW
import org.apache.hadoop.hbase.util.B
import org.apache.hadoop.io.LongW
import org.apache.hadoop.io.T
import org.apache.hadoop.mapreduce.M
* Mapper类，接收HDFS数据，写入到HBase表中
* @author fansy
public class ImportMapper extends Mapper{
private static final String COMMA = &,&;
private static final String COLON=&:&;
private String splitter =
// private String colsStr =
private int rkIndex =0; // rowkey 下标
private int tsIndex =1; // timestamp下标
private boolean hasTs = // 原始数据是否有timestamp
private SimpleDateFormat sf =
private ArrayList colsFamily=
private Put put =
ImmutableBytesWritable rowkey = new ImmutableBytesWritable();
protected void setup(Mapper.Context context)
throws IOException, InterruptedException {
splitter = context.getConfiguration().get(ImportToHBase.SPLITTER,&,&);
String colsStr = context.getConfiguration().get(ImportToHBase.COLSFAMILY,null);
sf = context.getConfiguration().get(ImportToHBase.DATEFORMAT,null)==null
? new SimpleDateFormat(&yyyy-MM-dd HH:mm&)
:new SimpleDateFormat(context.getConfiguration().get(ImportToHBase.DATEFORMAT));
String[] cols = colsStr.split(COMMA, -1);
colsFamily =new ArrayList&&();
for(int i=0;i& cols.i++){
if(&rk&.equals(cols[i])){
colsFamily.add(null);
if(&ts&.equals(cols[i])){
colsFamily.add(null);
hasTs = // 原始数据包括ts
colsFamily.add(getCol(cols[i]));
* 获取 family：qualifier byte数组
* @param col
private byte[][] getCol(String col) {
byte[][] fam_qua = new byte[2][];
String[] fam_quaStr = col.split(COLON, -1);
fam_qua[0]=
Bytes.toBytes(fam_quaStr[0]);
fam_qua[1]=
Bytes.toBytes(fam_quaStr[1]);
return fam_
protected void map(LongWritable key, Text value,
Mapper.Context context)
throws IOException, InterruptedException {
String[] words = value.toString().split(splitter, -1);
if(words.length!=colsFamily.size()){
System.out.println(&line:&+value.toString()+& does not compatible!&);
rowkey.set(getRowKey(words[rkIndex]));
put = getValue(words,colsFamily,rowkey.copyBytes());
context.write(rowkey, put);
* 获取Put值
* @param words
* @param colsFamily
* @param bs
private Put getValue(String[] words, ArrayList colsFamily, byte[] bs) {
Put put = new Put(bs);
for(int i=0;iMapper是整个流程的核心，主要负责进行数据解析、并从HDFS导入到HBase表中的工作，其各个部分功能如下：? setup()：获取输入数据字段分隔符，获取列簇、列名，获取rowkey列标，获取ts格式及列标（如果没有的话，就按照插入数据的时间设置）；? map()：解析、过滤并提取数据（需要的字段数据），生成Put对象，写入HBase；&6.3 针对records,user MR导入：&只需要进行拼凑参数，然后直接调用即可。
7. 实时数据加载
使用Java API来操作HBase数据库，完成实时HBase数据库更新，包括冠字号查询、存取款等功能。