oec数据挖掘用到的hbase

初始化数据配置参数接口；

public static DataSource createComboPooledDataSource(

HadJdbcConfig hadJdbcConfig) throws PropertyVetoException {

ComboPooledDataSource cpds = new ComboPooledDataSource();

try {

log.info("JdbcDriverClass:" + hadJdbcConfig.getDriverClass());

log.info("JdbcUrl:" + hadJdbcConfig.getJdbcUrl());

// 配置数据源

cpds.setDriverClass(hadJdbcConfig.getDriverClass());

cpds.setJdbcUrl(hadJdbcConfig.getJdbcUrl());

cpds.setUser(hadJdbcConfig.getUsername());

cpds.setPassword(hadJdbcConfig.getPassword());

cpds.setAcquireRetryAttempts(hadJdbcConfig.getAcquireRetryAttempts());

cpds.setAcquireIncrement(hadJdbcConfig.getAcquireIncrement());

cpds.setAcquireRetryDelay(hadJdbcConfig.getAcquireRetryDelay());

cpds.setAutoCommitOnClose(hadJdbcConfig.getAutoCommitOnClose());

cpds.setMinPoolSize(hadJdbcConfig.getMiniPoolSize());

cpds.setMaxPoolSize(hadJdbcConfig.getMaxPoolSize());

cpds.setInitialPoolSize(hadJdbcConfig.getInitialPoolSize());

cpds.setIdleConnectionTestPeriod(hadJdbcConfig.getIdleConnectionTestPeriod());

cpds.setTestConnectionOnCheckin(hadJdbcConfig.getTestConnectionOnCheckin());

cpds.setBreakAfterAcquireFailure(hadJdbcConfig.getBreakAfterAcquireFailure());

cpds.setMaxIdleTime(hadJdbcConfig.getMaxIdleTime());

log.info("Initialize dbcp initialization success ....");

} catch (PropertyVetoException e) {

e.printStackTrace();

log.fatal("[-1000][数据库连接故障][" + ExceptionUtil.getTrace(e) + "]");

throw e;

}

return (DataSource) cpds;

}

hbase查询和插入

/**

* 全表扫描（方案二）

*

* @param tablename

* @return

*/

public String scaner(String tablename) {

String rowKey; // 行健

String qualifier; // 列

String value; // 值

StringBuffer sbf = new StringBuffer();

boolean flag = true;

SqlDao sqlDao = new SqlDao();

try {

HTable table = new HTable(conf, tablename);

Scan s = new Scan();

s.setMaxVersions(1);

ResultScanner rs = table.getScanner(s);

//遍历所有行

for (Result r : rs) {

// 取出属性名

if (flag) {

int counter=0;

for (Cell cell : r.rawCells()) {

String rowkey = new String(CellUtil.cloneRow(cell))+"\t";

String family = new String(CellUtil.cloneFamily(cell));

String ruleId = new String(CellUtil.cloneQualifier(cell));

if(counter++<1){

sbf.append("rowkey"+"\t");

}

if(!ruleId.equals("data")){

//测试，不对ruleId进行转换

//qualifier = family+":"+ruleId+"\t";

//弃用

//qualifier = family+":"+sqlDao.queryByRuleId(ruleId) + "\t";

qualifier = sqlDao.queryByRuleId(ruleId) + "\t";

}else{

//弃用

//qualifier = family+":"+ruleId+"\t";

qualifier = sqlDao.queryByColumnId(family)+"\t";

}

sbf.append(qualifier);

}

sbf.append("\r\n");

}

// 取出一行的所有列值

int counter = 0;

for (Cell cell : r.rawCells()) {

if(counter++<1){

sbf.append(new String(CellUtil.cloneRow(cell))+"\t");

}

sbf.append(new String(CellUtil.cloneValue(cell)) + "\t");

}

sbf.append("\r\n");

flag = false;

}

} catch (IOException e) {

e.printStackTrace();

}

return new String(sbf);

}

/**

* 插入一行数据到HBase,返回true表示插入成功，返回false表示插入失败

* @param tableName

* @param rowkey

* @param columnFamily

* @param column

* @param value

* @return

*/

public static boolean insertOneRow(String tableName,String rowkey,String columnFamily,String column,String value){

HTable table;

try {

table = new HTable(conf, tableName);

Put put=new Put(Bytes.toBytes(rowkey));

put.add(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(value));

table.put(put);//放入表

table.close();//释放资源

return true;

} catch (IOException e) {

e.printStackTrace();

return false;

}

}