Akses LindormTable melalui Spark - Lindorm - Alibaba Cloud Documentation Center

Apache Spark dapat melakukan kueri dan menganalisis data yang disimpan di LindormTable melalui konektor HBase. Panduan ini memandu Anda melalui konfigurasi koneksi, pemetaan skema tabel HBase ke tabel Spark, serta menjalankan kueri Spark SQL.

Prasyarat

Sebelum memulai, pastikan Anda telah memiliki:

LindormTable versi 2.4.3 atau lebih baru
Alamat IP klien Anda telah ditambahkan ke daftar putih instans Lindorm. Lihat Konfigurasi daftar putih.
Titik akhir LindormTable dari API HBase untuk Java. Lihat Lihat titik akhir.

Catatan penggunaan

Untuk mengakses instans Lindorm melalui Internet atau jika instans Lindorm Anda merupakan instans single-node, lakukan upgrade SDK dan perbarui konfigurasi Anda sebelum melanjutkan. Lihat Langkah 1 dalam Gunakan API ApsaraDB for HBase untuk Java untuk terhubung ke dan menggunakan LindormTable.
Jika aplikasi Anda berjalan pada instans Elastic Compute Service (ECS), pastikan bahwa:
- Instans Lindorm dan instans ECS berada di wilayah yang sama. Sebaiknya deploy keduanya di zona yang sama untuk mengurangi latensi jaringan.
- Instans Lindorm dan instans ECS berada di VPC yang sama.

Alur kerja keseluruhan

Alur kerja keseluruhan untuk menghubungkan Spark ke LindormTable adalah:

Konfigurasikan titik akhir koneksi LindormTable.
Buat tabel HBase dan masukkan data.
Definisikan katalog untuk memetakan kolom HBase ke tabel Spark.
Buat tabel Spark menggunakan katalog tersebut.
Jalankan kueri Spark SQL terhadap tabel Spark.

Konfigurasi koneksi

Atur hbase.zookeeper.quorum ke titik akhir LindormTable dari API HBase untuk Java. Gunakan salah satu metode berikut:

Metode 1: File konfigurasi

Tambahkan berikut ini ke hbase-site.xml:

<configuration>
    <!-- Titik akhir LindormTable dari API HBase untuk Java -->
    <property>
        <name>hbase.zookeeper.quorum</name>
        <value>ld-m5ef25p66n5es****-proxy-lindorm.lindorm.rds.aliyuncs.com:30020</value>
    </property>
</configuration>

Metode 2: Objek konfigurasi

// Buat objek Configuration.
val conf = HBaseConfiguration.create()
// Tentukan titik akhir LindormTable dari API HBase untuk Java.
conf.set("hbase.zookeeper.quorum", "ld-m5ef25p66n5es****-proxy-lindorm.lindorm.rds.aliyuncs.com:30020")

Contoh

Contoh berikut menunjukkan alur kerja end-to-end secara lengkap: konfigurasi koneksi, pembuatan tabel HBase, pemasukan data, definisi tabel Spark dengan katalog, dan menjalankan kueri COUNT().

Langkah 1: Konfigurasi koneksi

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

Langkah 2: Buat tabel 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)

Langkah 3: Masukkan data

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)

Langkah 4: Buat tabel Spark menggunakan katalog

Katalog mendefinisikan pemetaan antara tabel HBase dan tabel Spark. Katalog tersebut menentukan kunci baris dan memetakan setiap kolom Spark ke keluarga kolom dan kualifikasi kolom HBase. Katalog diteruskan sebagai string JSON dalam klausa OPTIONS pernyataan 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)

Langkah 5: Jalankan kueri Spark SQL

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

Lindorm:Akses\ LindormTable\ melalui\ Spark