基于Mysql的表转HBase小Demo

/* * Copyright 1998-2012 360buy.com All right reserved. This software is the confidential and proprietary information of * 360buy.com ("Confidential Information"). You shall not disclose such Confidential Information and shall use it only * in accordance with the terms of the license agreement you entered into with 360buy.com. */ package com.ding.hbase.demo.HBaseDemo.dao; import java.io.IOException; import java.util.ArrayList; import java.util.List; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.KeyValue; import org.apache.hadoop.hbase.MasterNotRunningException; import org.apache.hadoop.hbase.ZooKeeperConnectionException; import org.apache.hadoop.hbase.client.Delete; import org.apache.hadoop.hbase.client.Get; import org.apache.hadoop.hbase.client.HBaseAdmin; import org.apache.hadoop.hbase.client.HTable; import org.apache.hadoop.hbase.client.HTablePool; import org.apache.hadoop.hbase.client.Put; import org.apache.hadoop.hbase.client.Result; import org.apache.hadoop.hbase.client.ResultScanner; import org.apache.hadoop.hbase.client.Scan; import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp; import org.apache.hadoop.hbase.filter.Filter; import org.apache.hadoop.hbase.filter.FilterList; import org.apache.hadoop.hbase.filter.SingleColumnValueFilter; import org.apache.hadoop.hbase.util.Bytes; import com.ding.hbase.demo.HBaseDemo.domain.UserHBase; /** * 类HBaseUtil.java的实现描述：TODO 类实现描述 * * @author gouding@360buy.com 2012-7-10 上午11:35:12 */ public class HBaseUtil { private static final String HBASE_ZOOKEEPER_PROPERTY_CLIENTPORT = "2222"; private static final String HBASE_ZOOKEEPER_QUORUM = "jss51"; /*----------------------------------------------------------------------------------------------*/ public static Configuration cfg = null; private static HBaseAdmin admin = null; private static HTablePool pool = null; static { cfg = new Configuration(); cfg.set("hbase.zookeeper.property.clientPort", HBASE_ZOOKEEPER_PROPERTY_CLIENTPORT); cfg.set("hbase.zookeeper.quorum", HBASE_ZOOKEEPER_QUORUM); cfg = HBaseConfiguration.create(cfg); try { admin = new HBaseAdmin(cfg); } catch (MasterNotRunningException e) { e.printStackTrace(); } catch (ZooKeeperConnectionException e) { e.printStackTrace(); } pool = new HTablePool(cfg, 15); } /** * 添加一条新记录 * * @param user * @param tableName * @param familys */ public static void addUser(UserHBase user, String tableName, String[] familys) { if (user.getUserName() != null) { writeRow(tableName, user.getUserName(), familys[0], "user_name", user.getUserName()); } if (user.getUserPassword() != null) { writeRow(tableName, user.getUserName(), familys[0], "password", user.getUserPassword()); } if (user.getUserEmail() != null) { writeRow(tableName, user.getUserName(), familys[0], "email", user.getUserEmail()); } if (user.getStatus() != null) { writeRow(tableName, user.getUserName(), familys[0], "state", user.getStatus()); } if (user.getNickName() != null) { writeRow(tableName, user.getUserName(), familys[1], "nick_name", user.getNickName()); } if (user.getTest() != null) { writeRow(tableName, user.getUserName(), familys[1], "test", user.getTest()); } } /** * 更新用户信息记录，更新时默认user对象中为null的属性是用户不想更改的属性 * * @param user * @param tableName * @param familys */ public static void updateUser(UserHBase user, String tableName, String[] familys) { if (user.getUserName() != null) { writeRow(tableName, user.getUserName(), familys[0], "user_name", user.getUserName()); } if (user.getUserPassword() != null) { writeRow(tableName, user.getUserName(), familys[0], "password", user.getUserPassword()); } if (user.getUserEmail() != null) { writeRow(tableName, user.getUserName(), familys[0], "email", user.getUserEmail()); } if (user.getStatus() != null) { writeRow(tableName, user.getUserName(), familys[0], "state", user.getStatus()); } if (user.getNickName() != null) { writeRow(tableName, user.getUserName(), familys[1], "nick_name", user.getNickName()); } if (user.getTest() != null) { writeRow(tableName, user.getUserName(), familys[1], "test", user.getTest()); } } /** * 删除一行记录 * * @param tablename * @param rowkey * @throws IOException */ public static void deleteRow(String tablename, String rowkey) throws IOException { if (!admin.tableExists(tablename)) { System.out.println("you don't hava the table " + tablename + "!"); return; } HTable table = getTableFromPool(tablename); List<Delete> list = new ArrayList<Delete>(); Delete d1 = new Delete(rowkey.getBytes()); list.add(d1); table.delete(list); System.out.println("删除行成功,该行的rowkey为： " + rowkey); } /** * 查找一行记录 * * @param tablename * @param rowkey */ public static UserHBase selectRow(String tablename, String rowKey) throws IOException { if (!admin.tableExists(tablename)) { System.out.println("you don't hava the table " + tablename + "!"); return null; } HTable table = getTableFromPool(tablename); Get g = new Get(rowKey.getBytes()); Result rs = table.get(g); UserHBase user = new UserHBase(); for (KeyValue value : rs.raw()) { String qualifierString = new String(value.getQualifier()); String valueString = new String(value.getValue()); if (qualifierString.equals("email")) { user.setUserEmail(valueString); } else if (qualifierString.equals("user_name")) { user.setUserName(valueString); } else if (qualifierString.equals("password")) { user.setUserPassword(valueString); } else if (qualifierString.equals("state")) { user.setStatus(valueString); } else if (qualifierString.equals("nick_name")) { user.setNickName(valueString); } else if (qualifierString.equals("test")) { user.setTest(valueString); } } return user; } /** * 查询表中某列值为指定值的所有数据 * * @param tablename * @param familyName * @param columnName * @param columnValue * @throws IOException */ public static UserHBase getRowsByColumnValue(String tableName, String familyName, String columnName, String columnValue) throws IOException { if (!admin.tableExists(tableName)) { System.out.println("you don't hava the table " + tableName + "!"); return null; } Filter filter = new SingleColumnValueFilter(familyName.getBytes(), columnName.getBytes(), CompareOp.EQUAL, columnValue.getBytes()); List<Filter> filters = new ArrayList<Filter>(); filters.add(filter); return getRowsByFilters(tabl