Spark no HBase com Spark shell

Alguns pequenos ajustes são necessários para atacar tabelas HBase de um contexto Spark.

Vamos primeiro criar rapidamente uma tabela de amostra "t1" HBase com 40 linhas

[root@sandbox ~]# cat hbase_load.txt
create 't1', 'f1'
for i in '1'..'10' do \
for j in '1'..'2' do \
for k in '1'..'2' do \
rnd=(0...64).map { (65 + rand(26)).chr }.join
put 't1', "#{i}-#{j}-#{k}", "f1:#{j}#{k}", "#{rnd}"
end \
end \
end
[root@sandbox ~]# cat hbase_load.txt |hbase shell

Você precisa ajustar seu caminho de classe Spark (guava 14 necessário, então incluí o primeiro que encontrei):

[root@sandbox ~]# export SPARK_CLASSPATH=/usr/hdp/current/spark-client/lib/hbase-common.jar:/usr/hdp/current/spark-client/lib/hbase-client.jar:/usr/hdp/current/spark-client/lib/hbase-protocol.jar:/usr/hdp/current/spark-client/lib/hbase-server.jar:/etc/hbase/conf:/usr/hdp/2.3.2.0-2950/oozie/share/lib/spark/guava-14.0.1.jar

[root@sandbox ~]# spark-shell --master yarn-client

Como uma observação lateral, o SPARK_CLASSPATH está obsoleto no Spark 1.5.x+, então você deve usar

[root@sandbox ~]# spark-shell --master yarn-client --driver-class-path=/usr/hdp/current/spark-client/lib/hbase-common.jar:/usr/hdp/current/spark-client/lib/hbase-client.jar:/usr/hdp/current/spark-client/lib/hbase-protocol.jar:/usr/hdp/current/spark-client/lib/hbase-hadoop2-compat.jar:/usr/hdp/current/spark-client/lib/hbase-server.jar:/etc/hbase/conf:/usr/hdp/2.3.2.0-2950/oozie/share/lib/spark/guava-14.0.1.jar

Eu encontrei bugs usando o anterior:[…]Causado por:java.lang.IllegalStateException:dados de bloco não lidos, então usei a primeira versão (usando SPARK_CLASSPATH) Agora é a vez de Scala !

scala>

import org.apache.hadoop.hbase.{HBaseConfiguration, HTableDescriptor}
import org.apache.hadoop.hbase.client.{HBaseAdmin, Result}
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.mapreduce.TableInputFormat

val tableName = "t1"
val hconf = HBaseConfiguration.create()
hconf.set(TableInputFormat.INPUT_TABLE, "t1")

val hBaseRDD = sc.newAPIHadoopRDD(hconf, classOf[TableInputFormat], classOf[ImmutableBytesWritable], classOf[Result])
println("records found : " + hBaseRDD.count())


[...]
2016-04-07 18:44:40,553 INFO [main] scheduler.DAGScheduler: Job 0 finished: count at <console>:30, took 2.092481 s
Number of Records found : 40

Se você quiser usar o HBase Admin para ver a lista de tabelas, snapshots ou qualquer operação relacionada ao administrador, use

scala> val admin = new HBaseAdmin(hconf)

admin.listTables

E se você quiser criar uma tabela:

val tableDesc = new HTableDescriptor(tableName)
admin.createTable(tableDesc)