nathanmarz/storm — Issue #51: java.io.NotSerializableException: org.apache.hadoop.conf.Configuration
Author: wangweichen
Created: 2011-11-22 09:00:45+00:00
State: open

hello!!
i want to use api of hbase in storm !!!
i create a class "hbaseOperation",code follow:

import java.io.IOException;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
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.Row;
import org.apache.hadoop.hbase.client.RowLock;
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.taobao.search.hbaserow.*;

public class hbaseOperation implements java.io.Serializable {

private Configuration cfg;
private int pool_size;
private HTablePool pool;

public hbaseOperation(){
cfg = HBaseConfiguration.create();
cfg.set("hbase.zookeeper.property.clientPort", "2181");
this.cfg.set("hbase.zookeeper.quorum",
);
this.cfg.set("zookeeper.znode.parent", "/hbase");
this.pool_size = 1000;
this.pool = new HTablePool(this.cfg, this.pool_size);
}

public hbaseOperation(int pool_size){
this.cfg = HBaseConfiguration.create();
this.cfg.set("hbase.zookeeper.property.clientPort", "2181");
this.cfg.set("hbase.zookeeper.quorum",
);
this.cfg.set("zookeeper.znode.parent", "/hbase");
this.pool_size = pool_size;
this.pool = new HTablePool(this.cfg, pool_size);
}

public RowOperation queryByRowKey(String tableName, String row_key) {
HTable table = (HTable) this.pool.getTable(tableName);
try {
Get scan = new Get(row_key.getBytes());
Result r = table.get(scan);
//Row row = new Row(new String(r.getRow()));
//Row row = new Row(new )
//Row row = new Row();
//Row row = new Row(new String(r.getRow()));
RowOperation row = new RowOperation(new String(r.getRow()));
for (KeyValue keyValue : r.raw()) {
row.addClomn(new String(keyValue.getFamily()), new String(
keyValue.getValue()));
}
return row;
} catch (IOException e) {
e.printStackTrace();
} finally {
this.pool.putTable(table);
}
return null;
}

public void addData(String tableName, RowOperation row) throws Exception {
// 从HTablePool中得到table而不是单独新建
HTable table = (HTable) this.pool.getTable(tableName);
// 结果达到一定数量在写入数据库
table.setAutoFlush(false);
Put p = new Put(row.row_key.getBytes());
// 在正式运行时要设置为true确保数据正确写入
p.setWriteToWAL(false);

for (Map.Entry entry : row.colums.entrySet())
p.add(((String) entry.getKey()).getBytes(), null,
((String) entry.getValue()).getBytes());
try {
table.put(p);

table.flushCommits();
} catch (IOException e) {
e.printStackTrace();
} finally {
this.pool.putTable(table);
}
}

public void queryByKeyUpdate(String tableName,String row_key,String column,String value) throws Exception{
HTable table = (HTable) this.pool.getTable(tableName);
RowLock rowlock = table.lockRow(row_key.getBytes());
try{
// Take a lock, no one else can do something with this row while we have the lock
////RowLock rowLock = htable.lockRow(rowkey);

Get get = new Get(row_key.getBytes(),rowlock);
Result res = table.get(get);
//if (!res.isEmpty()) {
// throw new Exception("Row with this key already exists.");
//}
RowOperation row = new RowOperation(new String(res.getRow()));
for(KeyValue keyValue : res.raw()){
//row.addClomn(new String(keyValue.getFamily()), new String(keyValue.getValue()));
//if (new String(keyValue.getFamily()) == column)
if(new String(keyValue.getFamily()).equals(column)){
row.addClomn(new String(keyValue.getFamily()), value);
}
else{
row.addClomn(new String(keyValue.getFamily()), new String(keyValue.getValue()));
}
}

addData(tableName,row);

}catch (IOException e) {
e.printStackTrace();
} finally {
this.pool.putTable(table);
table.unlockRow(rowlock);
}
//return null;
}
}

but , when i run jar on storm cluster !!!!!
has error:
Exception in thread "main" java.lang.RuntimeException: java.io.NotSerializableException: org.apache.hadoop.conf.Configuration
at backtype.storm.utils.Utils.serialize(Utils.java:35)
at backtype.storm.topology.TopologyBuilder.createTopology(TopologyBuilder.java:87)
at com.taobao.search.enter.stormRun.main(stormRun.java:22)
Caused by: java.io.NotSerializableException: org.apache.hadoop.conf.Configuration
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1156)
at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1509)
at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1474)
at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1392)
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1150)
at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1509)
at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1474)
at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1392)
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1150)
at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1509)
at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1474)
at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1392)
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1150)
at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:326)
at backtype.storm.utils.Utils.serialize(Utils.java:31)
... 2 more

i do not know!! how to resolve this problem！！！！