in spark/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/example/rdd/HBaseBulkGetExample.scala [38:100]
def main(args: Array[String]) {
if (args.length < 1) {
println("HBaseBulkGetExample {tableName} is missing an argument")
return
}
val tableName = args(0)
val sparkConf = new SparkConf().setAppName("HBaseBulkGetExample " + tableName)
val sc = new SparkContext(sparkConf)
try {
// [(Array[Byte])]
val rdd = sc.parallelize(
Array(
Bytes.toBytes("1"),
Bytes.toBytes("2"),
Bytes.toBytes("3"),
Bytes.toBytes("4"),
Bytes.toBytes("5"),
Bytes.toBytes("6"),
Bytes.toBytes("7")))
val conf = HBaseConfiguration.create()
val hbaseContext = new HBaseContext(sc, conf)
val getRdd = rdd.hbaseBulkGet[String](
hbaseContext,
TableName.valueOf(tableName),
2,
record => {
System.out.println("making Get")
new Get(record)
},
(result: Result) => {
val it = result.listCells().iterator()
val b = new StringBuilder
b.append(Bytes.toString(result.getRow) + ":")
while (it.hasNext) {
val cell = it.next()
val q = Bytes.toString(CellUtil.cloneQualifier(cell))
if (q.equals("counter")) {
b.append("(" + q + "," + Bytes.toLong(CellUtil.cloneValue(cell)) + ")")
} else {
b.append("(" + q + "," + Bytes.toString(CellUtil.cloneValue(cell)) + ")")
}
}
b.toString()
})
getRdd
.collect()
.foreach(v => println(v))
} finally {
sc.stop()
}
}