Lindorm:Accéder à LindormTable via Spark

Prérequis

Avant de commencer, assurez-vous de disposer des éléments suivants :

• LindormTable version 2.4.3 ou ultérieure

• L'adresse IP de votre client ajoutée à la liste d'autorisation de l'instance Lindorm. Consultez Configurer les listes d'autorisation

• L'endpoint LindormTable de l'API HBase pour Java. Consultez Afficher les endpoints

Notes d'utilisation

• Pour accéder à une instance Lindorm via Internet, ou si votre instance Lindorm est une instance à nœud unique, mettez à jour votre SDK et actualisez vos configurations avant de poursuivre. Reportez-vous à l'étape 1 de la section Utiliser l'API ApsaraDB for HBase pour Java afin de se connecter à LindormTable et de l'utiliser.

• Si votre application s'exécute sur une instance Elastic Compute Service (ECS), vérifiez que :

  • Votre instance Lindorm et votre instance ECS se trouvent dans la même région. Déployez-les dans la même zone pour réduire la latence réseau.

  • Votre instance Lindorm et votre instance ECS partagent le même VPC.

Processus global

La procédure générale pour connecter Spark à LindormTable est la suivante :

1. Configurez l'endpoint de connexion LindormTable.

2. Créez une table HBase et insérez des données.

3. Définissez un catalogue pour mapper les colonnes HBase vers une table Spark.

4. Créez la table Spark à l'aide du catalogue.

5. Exécutez des requêtes Spark SQL sur la table Spark.

Configurer la connexion

Définissez hbase.zookeeper.quorum avec l'endpoint LindormTable de l'API HBase pour Java. Utilisez l'une des méthodes suivantes :

Méthode 1 : Fichier de configuration

Ajoutez les lignes suivantes au fichier 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éthode 2 : Objet de configuration

// 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")

Exemple

L'exemple suivant illustre le flux de travail complet de bout en bout : configurer la connexion, créer une table HBase, insérer des données, définir une table Spark avec un catalogue et exécuter une requête COUNT().

var conf = HBaseConfiguration.create
conf.set("hbase.zookeeper.quorum", "ld-m5ef25p66n5es****-proxy-lindorm.lindorm.rds.aliyuncs.com:30020")

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)

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)

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)

val result = sparkSession.sql("select count(*) from " + sparkTableName)
val sparkCounts = result.collect().apply(0).getLong(0)
println("Row count: " + sparkCounts)

Étapes suivantes

• Utiliser l'API ApsaraDB for HBase pour Java afin de se connecter à LindormTable et de l'utiliser

• Afficher les endpoints