通过Spark访问宽表引擎

本文介绍如何通过Spark以开源的方式访问Lindorm宽表。

前提条件

  • 宽表引擎为2.4.3及以上版本。

  • 已将客户端IP地址添加至Lindorm白名单。如何添加,请参见设置白名单

  • 已获取宽表引擎的HBase Java API连接地址。如何获取,请参见查看连接地址

注意事项

  • 如果您想要通过公网访问或您的实例类型为Lindorm单节点,在执行本文操作前,需要先升级SDK并更改配置。具体操作,请参见通过HBase Java API连接并使用宽表引擎章节中的步骤1。

  • 如果应用部署在ECS实例,通过专有网络访问Lindorm实例前,需要确保Lindorm实例和ECS实例满足以下条件,以保证网络的连通性。

    • 所在地域相同,并建议所在可用区相同(以减少网络延时)。

    • ECS实例与Lindorm实例属于同一专有网络。

添加Lindorm访问配置

  • 方式一:通过配置文件添加访问配置。

    在配置文件hbase-site.xml中增加下列配置项:

    <configuration>
          <!--
        宽表引擎的HBase Java API连接地址
        -->
        <property>
            <name>hbase.zookeeper.quorum</name>
            <value>ld-m5ef25p66n5es****-proxy-lindorm.lindorm.rds.aliyuncs.com:30020</value>
        </property>
    </configuration>
  • 方式二:通过代码在Configuration中添加参数。

    // 新建一个Configuration
    Configuration conf = HBaseConfiguration.create();
    // 宽表引擎的HBase Java API连接地址
    conf.set("hbase.zookeeper.quorum", "ld-m5ef25p66n5es****-proxy-lindorm.lindorm.rds.aliyuncs.com:30020");

Spark访问示例

test(" test the spark sql count result") {
  //1. 添加HBaseue访问配置
  var conf = HBaseConfiguration.create
  conf.set("hbase.zookeeper.quorum", "ld-m5ef25p66n5es****-proxy-lindorm.lindorm.rds.aliyuncs.com:30020")
  //2. 创建表
  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)
  //3. 插入测试数据
  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)
  //4. 创建spark表
  val sparkTableName = "spark_hbase"
  val createCmd = s"""CREATE TABLE ${sparkTableName} USING org.apache.hadoop.hbase.spark
                         |    OPTIONS ('catalog'=
                         |    '{"table":{"namespace":"$${hbaseTableName}",                   "name":"${hbaseTableName}"},"rowkey":"rowkey",
                         |    "columns":{
                         |    "col0":{"cf":"rowkey", "col":"rowkey", "type":"string"},
                         |    "col1":{"cf":"cf1", "col":"a", "type":"string"},
                         |    "col2":{"cf":"cf1", "col":"b", "type":"String"}}}'
                         |    )""".stripMargin
  println(" createCmd: \n" + createCmd + " rows : " + rowsCount)
  sparkSession.sql(createCmd)
  //5. 执行count sql
  val result = sparkSession.sql("select count(*) from " + sparkTableName)
  val sparkCounts = result.collect().apply(0).getLong(0)
  println(" sparkCounts : " + sparkCounts)