O Apache Spark permite consultar e analisar dados armazenados no LindormTable usando o conector HBase. Este guia orienta você na configuração da conexão, no mapeamento do esquema da tabela HBase para uma tabela Spark e na execução de consultas Spark SQL.
Pré-requisitos
Antes de começar, verifique se você possui:
LindormTable versão 2.4.3 ou posterior
Endereço IP do cliente adicionado à lista de permissões da instância Lindorm. Consulte Configurar listas de permissões
Endpoint do LindormTable para a API HBase para Java. Consulte Visualizar endpoints
Observações de uso
Para acessar uma instância Lindorm pela Internet, ou caso sua instância seja de nó único, atualize o SDK e as configurações antes de prosseguir. Consulte a Etapa 1 em Usar a API ApsaraDB for HBase para Java para conectar-se e usar o LindormTable.
-
Se a aplicação estiver executando em uma instância do Elastic Compute Service (ECS), garanta que:
As instâncias Lindorm e ECS estejam na mesma região. Implante-as na mesma zona para reduzir a latência de rede.
As instâncias Lindorm e ECS estejam na mesma VPC.
Processo geral
O fluxo completo para conectar o Spark ao LindormTable consiste em:
Configure o endpoint de conexão do LindormTable.
Crie uma tabela HBase e insira dados.
Defina um catálogo para mapear colunas HBase para uma tabela Spark.
Crie a tabela Spark utilizando o catálogo.
Execute consultas Spark SQL na tabela Spark.
Configure a conexão
Defina hbase.zookeeper.quorum como o endpoint do LindormTable para a API HBase para Java. Utilize um dos métodos abaixo:
Método 1: Arquivo de configuração
Adicione o seguinte conteúdo ao arquivo hbase-site.xml:
<configuration>
<!-- LindormTable endpoint of the HBase API for Java -->
<property>
<name>hbase.zookeeper.quorum</name>
<value>ld-m5ef25p66n5es****-proxy-lindorm.lindorm.rds.aliyuncs.com:30020</value>
</property>
</configuration>Método 2: Objeto de configuração
// Create a Configuration object.
val conf = HBaseConfiguration.create()
// Specify the LindormTable endpoint of the HBase API for Java.
conf.set("hbase.zookeeper.quorum", "ld-m5ef25p66n5es****-proxy-lindorm.lindorm.rds.aliyuncs.com:30020")Exemplo
O exemplo a seguir demonstra o fluxo completo de ponta a ponta: configure a conexão, crie uma tabela HBase, insira dados, defina uma tabela Spark com catálogo e execute uma consulta COUNT().
Etapa 1: Configure a conexão
var conf = HBaseConfiguration.create
conf.set("hbase.zookeeper.quorum", "ld-m5ef25p66n5es****-proxy-lindorm.lindorm.rds.aliyuncs.com:30020")Etapa 2: Crie uma tabela HBase
val hbaseTableName = "testTable"
val cf = "f"
val column1 = cf + ":a"
val column2 = cf + ":b"
var namespace = "spark_test"
val admin = ConnectionFactory.createConnection(conf).getAdmin()
val tableName = TableName.valueOf(namespace, hbaseTableName)
val htd = new HTableDescriptor(tableName)
htd.addFamily(new HColumnDescriptor(cf))
admin.createTable(htd)Etapa 3: Inserir dados
val k: Array[Byte] = new Array[Byte](3)
val famAndQf = KeyValue.parseColumn(Bytes.toBytes(column1))
val puts = new util.ArrayList[Put]()
var i = 0
for (b1 <- ('a' to 'z')) {
for (b2 <- ('a' to 'z')) {
for (b3 <- ('a' to 'z')) {
if (i < 10) {
k(0) = b1.toByte
k(1) = b2.toByte
k(2) = b3.toByte
val put = new Put(k)
put.addColumn(famAndQf(0), famAndQf(1), ("value_" + b1 + b2 + b3).getBytes())
puts.add(put)
i = i + 1
}
}
}
}
val conn = ConnectionFactory.createConnection(conf)
val table = conn.getTable(tableName)
table.put(puts)Etapa 4: Crie uma tabela Spark usando um catálogo
Um catálogo define o mapeamento entre a tabela HBase e uma tabela Spark. Ele especifica a chave de linha e associa cada coluna Spark a uma família de colunas e a um qualificador de coluna HBase. O catálogo é fornecido como uma string JSON na cláusula OPTIONS da instrução CREATE TABLE.
val sparkTableName = "spark_hbase"
val createCmd =
s"""CREATE TABLE ${sparkTableName} USING org.apache.hadoop.hbase.spark
|OPTIONS ('catalog'=
|'{"table":{"namespace":"${namespace}", "name":"${hbaseTableName}"},
|"rowkey":"rowkey",
|"columns":{
|"col0":{"cf":"rowkey", "col":"rowkey", "type":"string"},
|"col1":{"cf":"f", "col":"a", "type":"string"},
|"col2":{"cf":"f", "col":"b", "type":"string"}}}')
""".stripMargin
sparkSession.sql(createCmd)Etapa 5: Execute uma consulta Spark SQL
val result = sparkSession.sql("select count(*) from " + sparkTableName)
val sparkCounts = result.collect().apply(0).getLong(0)
println("Row count: " + sparkCounts)