E-MapReduce:Baca dari dan tulis ke HBase

Langkah 1: Dapatkan file JAR HBase Spark connector

Dapatkan paket dependensi yang diperlukan dengan mengikuti langkah-langkah berikut. Untuk kompatibilitas versi antara Spark, Scala, Hadoop, dan HBase, lihat dokumentasi resmi HBase Spark connector.

1. Kompilasi dan buat paket connector.

   Gunakan versi Spark, Scala, Hadoop, dan HBase dari lingkungan Anda untuk mengompilasi HBase Spark connector. Proses ini menghasilkan dua file JAR inti berikut:

   • hbase-spark-1.1.0-SNAPSHOT.jar

   • hbase-spark-protocol-shaded-1.1.0-SNAPSHOT.jar

     Sebagai contoh, gunakan perintah Maven berikut untuk mengompilasi dan membuat paket connector dengan versi tertentu.

     mvn -Dspark.version=3.4.2 -Dscala.version=2.12.10 -Dhadoop-three.version=3.2.0 -Dscala.binary.version=2.12 -Dhbase.version=2.4.9 clean package -DskipTests

     Jika lingkungan Anda menggunakan versi yang sama (Spark 3.4.2, Scala 2.12.10, Hadoop 3.2.0, dan HBase 2.4.9), Anda dapat menggunakan file JAR prakompilasi berikut:

     • hbase-spark-1.1.0-SNAPSHOT.jar

     • hbase-spark-protocol-shaded-1.1.0-SNAPSHOT.jar

2. Dapatkan dependensi HBase yang diperlukan. Ekstrak paket dependensi berikut dari folder lib/shaded-clients dan lib/client-facing-thirdparty di direktori instalasi HBase, dengan 2.4.9 sebagai nomor versi HBase.

   • hbase-shaded-client-2.4.9.jar

   • hbase-shaded-mapreduce-2.4.9.jar

   • slf4j-log4j12-1.7.30.jar

3. Unggah kelima file JAR tersebut ke OSS. Lihat Simple upload untuk instruksinya.

Langkah 2: Buat koneksi jaringan

EMR Serverless Spark memerlukan koneksi jaringan untuk mengakses layanan HBase di kluster HBase. Untuk informasi selengkapnya tentang koneksi jaringan, lihat Connect EMR Serverless Spark to other VPCs.

Langkah 3: Buat tabel HBase

1. Lakukan koneksi ke kluster menggunakan SSH. Untuk detailnya, lihat Log on to a cluster.

2. Jalankan perintah berikut untuk terhubung ke HBase.

   hbase shell

3. Jalankan perintah berikut untuk membuat tabel uji.

   create 'hbase_table', 'c1', 'c2'

4. Jalankan perintah berikut untuk menulis data uji ke tabel.

   put 'hbase_table', 'r1', 'c1:name', 'Alice'
   put 'hbase_table', 'r1', 'c1:age', '25'
   put 'hbase_table', 'r1', 'c2:city', 'New York'
   put 'hbase_table', 'r2', 'c1:name', 'Bob'
   put 'hbase_table', 'r2', 'c1:age', '30'
   put 'hbase_table', 'r2', 'c2:city', 'San Francisco'

Langkah 4: Baca tabel HBase

1. Buat session notebook. Untuk detailnya, lihat Manage notebook sessions.

   Saat membuat session, pilih versi engine yang sesuai dengan HBase Spark connector Anda dari daftar drop-down Engine Version. Pada bagian Normal Network Connection, pilih koneksi jaringan yang telah Anda buat di Langkah 2. Kemudian, pada bagian Spark Configuration, tambahkan parameter berikut untuk memuat HBase Spark connector.

   spark.jars                                        oss://<bucketname>/path/to/hbase-shaded-client-2.4.9.jar,oss://<bucketname>/path/to/hbase-shaded-mapreduce-2.4.9.jar,oss://<bucketname>/path/to/hbase-spark-1.1.0-SNAPSHOT.jar,oss://<bucketname>/path/to/hbase-spark-protocol-shaded-1.1.0-SNAPSHOT.jar,oss://<bucketname>/path/to/slf4j-log4j12-1.7.30.jar
   spark.hadoop.hbase.zookeeper.quorum               Zookeeper internal IP address
   spark.hadoop.hbase.zookeeper.property.clientPort  Zookeeper service port

   Tabel berikut menjelaskan parameter-parameter tersebut.

   Parameter	Deskripsi	Contoh
   spark.jars	Jalur OSS ke file JAR dependensi eksternal.	Sebagai contoh, salah satu dari lima file yang diunggah ke OSS adalah oss://<yourBucketname>/spark/hbase/hbase-shaded-client-2.4.9.jar.
   spark.hadoop.hbase.zookeeper.quorum	Alamat IP internal quorum ZooKeeper.	Jika Anda menggunakan kluster HBase yang berbeda, tentukan konfigurasi yang sesuai. Jika Anda menggunakan kluster EMR HBase, Anda dapat menemukan Private IP Address dari node master pada tab Nodes kluster.
   spark.hadoop.hbase.zookeeper.property.clientPort	Port layanan ZooKeeper.	Jika Anda menggunakan kluster HBase yang berbeda, tentukan port yang sesuai. Jika Anda menggunakan kluster EMR HBase Alibaba Cloud, port-nya adalah 2181.

2. Pada halaman Development, buat task bertipe Interactive Development > Notebook. Lalu, di pojok kanan atas, pilih session notebook yang telah Anda buat.

   Untuk informasi selengkapnya, lihat Manage notebook sessions.

3. Salin kode berikut ke dalam cell baru di notebook, sesuaikan parameter jika diperlukan, lalu klik Run.

   # Baca data dari tabel HBase.
   df = spark.read.format("org.apache.hadoop.hbase.spark") \
       .option("hbase.columns.mapping", "id STRING :key, name STRING c1:name, age STRING c1:age, city STRING c2:city") \
       .option("hbase.table", "hbase_table") \
       .option("hbase.spark.pushdown.columnfilter", False) \
       .load()
   # Daftarkan tampilan temporary.
   df.createOrReplaceTempView("hbase_table_view")
   # Kueri data menggunakan SQL.
   results = spark.sql("SELECT * FROM hbase_table_view")
   results.show()
   

   Jika data yang diharapkan dikembalikan, konfigurasi Anda sudah benar.

Langkah 5: Tulis ke tabel HBase

Salin kode berikut ke dalam cell baru di notebook, sesuaikan parameter jika diperlukan, lalu klik Run.

from pyspark.sql.types import StructType, StructField, StringType
data = [
    ("r3", "sam", "26", "New York")
]
schema = StructType([
    StructField("id", StringType(), True),
    StructField("name", StringType(), True),
    StructField("age", StringType(), True),
    StructField("city", StringType(), True)
])
testDS = spark.createDataFrame(data=data,schema=schema)
testDS.write.format("org.apache.hadoop.hbase.spark").option("hbase.columns.mapping", "id STRING :key, name STRING c1:name, age STRING c1:age, city STRING c2:city").option("hbase.table", "hbase_table").save()

Setelah menulis data, kueri tabel untuk memverifikasi bahwa operasi berhasil.

df = spark.read.format("org.apache.hadoop.hbase.spark") \
    .option("hbase.columns.mapping", "id STRING :key, name STRING c1:name, age STRING c1:age, city STRING c2:city") \
    .option("hbase.table", "hbase_table") \
    .option("hbase.spark.pushdown.columnfilter", False) \
    .load()
df.createOrReplaceTempView("hbase_table_view")
results = spark.sql("SELECT * FROM hbase_table_view")
results.show()