ApsaraDB for HBase:Use Spark to access an ApsaraDB for HBase Performance-enhanced Edition cluster

Prerequisites

Before you begin, make sure you have:

• An ApsaraDB for HBase Performance-enhanced Edition cluster running version 2.4.3 or later. To check or update the version, see Minor version updates.

• The client IP address added to the cluster allowlist. See Configure IP address allowlists and security groups.

• The Java API endpoint of the cluster, available on the Database Connection page in the ApsaraDB for HBase console.

Usage notes

• Internet access: Replace the open source HBase client with the ApsaraDB for HBase client before accessing the cluster over the internet. See Upgrade ApsaraDB for HBase SDK for Java.

• VPC access from an Elastic Compute Service (ECS) instance: Make sure the cluster and the ECS instance meet both of the following conditions:

  • They are deployed in the same region. Deploy them in the same zone to reduce network latency.

  • They belong to the same virtual private cloud (VPC).

Overall process

To use Spark SQL to query data in an ApsaraDB for HBase Performance-enhanced Edition cluster:

1. Configure the cluster connection — set the hbase.zookeeper.quorum property to the Java API endpoint.

2. Create an HBase table and insert test data.

3. Define a catalog JSON that maps the Spark schema to the HBase table.

4. Create a Spark table using the catalog definition.

5. Run Spark SQL queries against the mapped Spark table.

Configure the connection

Set the hbase.zookeeper.quorum property to the Java API endpoint of your cluster. Use either of the following methods.

Method 1: Configuration file

Add the following to hbase-site.xml:

<configuration>
    <!--
    The Java API endpoint of the cluster.
    Get the endpoint on the Database Connection page in the ApsaraDB for HBase console.
    -->
    <property>
        <name>hbase.zookeeper.quorum</name>
        <value>ld-bp150tns0sjxs****-proxy-hbaseue.hbaseue.rds.aliyuncs.com:30020</value>
    </property>
</configuration>

Method 2: Configuration object

Set the property programmatically in a Configuration object:

// Create a Configuration object.
Configuration conf = HBaseConfiguration.create();
// The Java API endpoint of the cluster.
// Get the endpoint on the Database Connection page in the ApsaraDB for HBase console.
conf.set("hbase.zookeeper.quorum", "ld-bp150tns0sjxs****-proxy-hbaseue.hbaseue.rds.aliyuncs.com:30020");

Example

The following example creates an HBase table, inserts test data, maps the table to a Spark SQL table using a catalog JSON definition, and runs a COUNT(*) query.

HTableDescriptor and HColumnDescriptor are deprecated in Apache HBase 2.x. Use TableDescriptorBuilder and ColumnFamilyDescriptorBuilder in production code.

test(" test the spark sql count result") {
  // Step 1: Configure the connection.
  var conf = HBaseConfiguration.create
  conf.set("hbase.zookeeper.quorum", "ld-bp150tns0sjxs****-proxy-hbaseue.hbaseue.rds.aliyuncs.com:30020")

  // Step 2: Create an HBase table.
  val hbaseTableName = "testTable"
  val cf = "f"
  val column1 = cf + ":a"
  val column2 = cf + ":b"
  var rowsCount: Int = -1
  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)

  // Step 3: Insert test data.
  val rng = new Random()
  val k: Array[Byte] = new Array[Byte](3)
  val famAndQf = KeyValue.parseColumn(Bytes.toBytes(column))
  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)

  // Step 4: Create a Spark table using a catalog JSON definition.
  // The catalog maps Spark column names to HBase column families and qualifiers.
  // - "table": specifies the HBase namespace and table name.
  // - "rowkey": identifies which Spark column maps to the HBase row key.
  // - "columns": maps each Spark column to an HBase column family ("cf"),
  //              column qualifier ("col"), and data type ("type").
  //              The row key is also listed here as a named column with cf "rowkey".
  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
  println("createCmd: \n" + createCmd + " rows : " + rowsCount)
  sparkSession.sql(createCmd)

  // Step 5: Run a Spark SQL COUNT(*) query.
  val result = sparkSession.sql("select count(*) from " + sparkTableName)
  val sparkCounts = result.collect().apply(0).getLong(0)
  println("sparkCounts : " + sparkCounts)
}

What to do next

• To access the cluster using the Java API directly, see Upgrade ApsaraDB for HBase SDK for Java.

• To manage cluster versions, see Minor version updates.