spark to hbase

[jinhong lu]

Hi, 

I write my result to hdfs, it did well:

val model = 
lines.map(pairFunction).groupByKey().flatMap(pairFlatMapFunction).aggregateByKey(new
 TrainFeature())(seqOp, combOp).values
 model.map(a => (a.toKey() + "\t" + a.totalCount + "\t" + 
a.positiveCount)).saveAsTextFile(modelDataPath);

But when I want to write to hbase, the applicaton hung, no log, no response, 
just stay there, and nothing is written to hbase:

val model = 
lines.map(pairFunction).groupByKey().flatMap(pairFlatMapFunction).aggregateByKey(new
 TrainFeature())(seqOp, combOp).values.foreach({ res =>
  val configuration = HBaseConfiguration.create();
  configuration.set("hbase.zookeeper.property.clientPort", "2181");
  configuration.set("hbase.zookeeper.quorum", “192.168.1.66");
  configuration.set("hbase.master", "192.168.1:60000");
  val hadmin = new HBaseAdmin(configuration);
  val table = new HTable(configuration, "ljh_test3");
  var put = new Put(Bytes.toBytes(res.toKey()));
  put.add(Bytes.toBytes("f"), Bytes.toBytes("c"), Bytes.toBytes(res.totalCount 
+ res.positiveCount));
  table.put(put);
  table.flushCommits()
})

And then I try to write som simple data to hbase, it did well too:

sc.parallelize(Array(1,2,3,4)).foreach({ res =>
val configuration = HBaseConfiguration.create();
configuration.set("hbase.zookeeper.property.clientPort", "2181");
configuration.set("hbase.zookeeper.quorum", "192.168.1.66");
configuration.set("hbase.master", "192.168.1:60000");
val hadmin = new HBaseAdmin(configuration);
val table = new HTable(configuration, "ljh_test3");
var put = new Put(Bytes.toBytes(res));
put.add(Bytes.toBytes("f"), Bytes.toBytes("c"), Bytes.toBytes(res));
table.put(put);
table.flushCommits()
})

what is the problem with the 2rd code? thanks a lot.