Menggunakan Apache Spark dengan HBase di Dataproc

Tujuan

Tutorial ini menunjukkan kepada Anda cara:

  1. Buat cluster Dataproc, instal Apache HBase dan Apache ZooKeeper di cluster
  2. Buat tabel HBase menggunakan shell HBase yang berjalan di node master cluster Dataproc
  3. Gunakan Cloud Shell untuk mengirimkan tugas Spark Java atau PySpark ke layanan Dataproc yang menulis data ke, lalu membaca data dari, tabel HBase

Biaya

Dalam dokumen ini, Anda akan menggunakan komponen Google Cloudyang dapat ditagih berikut:

Untuk membuat perkiraan biaya berdasarkan proyeksi penggunaan Anda, gunakan kalkulator harga.

Pengguna Google Cloud baru mungkin memenuhi syarat untuk mendapatkan uji coba gratis.

Sebelum memulai

Jika Anda belum melakukannya, buat project Google Cloud Platform.

  1. Sign in to your Google Cloud account. If you're new to Google Cloud, create an account to evaluate how our products perform in real-world scenarios. New customers also get $300 in free credits to run, test, and deploy workloads.
  2. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

    Go to project selector

  3. Verify that billing is enabled for your Google Cloud project.

  4. Enable the Dataproc and Compute Engine APIs.

    Enable the APIs

  5. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

    Go to project selector

  6. Verify that billing is enabled for your Google Cloud project.

  7. Enable the Dataproc and Compute Engine APIs.

    Enable the APIs

  8. Membuat cluster Dataproc

    1. Jalankan perintah berikut di terminal sesi Cloud Shell untuk:

      • Instal komponen HBase dan ZooKeeper
      • Sediakan tiga node pekerja (tiga hingga lima pekerja direkomendasikan untuk menjalankan kode dalam tutorial ini)
      • Aktifkan Component Gateway
      • Menggunakan versi gambar 2.0
      • Gunakan tanda --properties untuk menambahkan konfigurasi HBase dan library HBase ke jalur class driver dan eksekutor Spark.
    gcloud dataproc clusters create cluster-name \
        --region=region \
        --optional-components=HBASE,ZOOKEEPER \
        --num-workers=3 \
        --enable-component-gateway \
        --image-version=2.0 \
        --properties='spark:spark.driver.extraClassPath=/etc/hbase/conf:/usr/lib/hbase/*,spark:spark.executor.extraClassPath=/etc/hbase/conf:/usr/lib/hbase/*'
    

    Memverifikasi penginstalan konektor

    1. Dari Google Cloud konsol atau terminal sesi Cloud Shell, SSH ke node master cluster Dataproc.

    2. Verifikasi penginstalan konektor Apache HBase Spark di node master:

      ls -l /usr/lib/spark/jars | grep hbase-spark
      
      Contoh output:
      -rw-r--r-- 1 root root size date time hbase-spark-connector.version.jar
      

    3. Biarkan terminal sesi SSH tetap terbuka untuk:

      1. Membuat tabel HBase
      2. (Pengguna Java): jalankan perintah di node master cluster untuk menentukan versi komponen yang diinstal di cluster
      3. Pindai tabel Hbase setelah Anda menjalankan kode

    Membuat tabel HBase

    Jalankan perintah yang tercantum di bagian ini di terminal sesi SSH node master yang Anda buka pada langkah sebelumnya.

    1. Buka shell HBase:

      hbase shell
      

    2. Buat 'my-table' HBase dengan grup kolom 'cf':

      create 'my_table','cf'
      

      1. Untuk mengonfirmasi pembuatan tabel, di konsol Google Cloud , klik HBase di link Component Gateway konsolGoogle Cloud untuk membuka UI Apache HBase. my-table tercantum di bagian Tabel di halaman Beranda.

    Melihat kode Spark

    Java

    package hbase;
    
    import org.apache.hadoop.hbase.spark.datasources.HBaseTableCatalog;
    import org.apache.spark.sql.Dataset;
    import org.apache.spark.sql.Row;
    import org.apache.spark.sql.SparkSession;
    
    import java.io.Serializable;
    import java.util.Arrays;
    import java.util.HashMap;
    import java.util.Map;
    
    public class SparkHBaseMain {
        public static class SampleData implements Serializable {
            private String key;
            private String name;
    
    
            public SampleData(String key, String name) {
                this.key = key;
                this.name = name;
            }
    
            public SampleData() {
            }
    
            public String getName() {
                return name;
            }
    
            public void setName(String name) {
                this.name = name;
            }
    
            public String getKey() {
                return key;
            }
    
            public void setKey(String key) {
                this.key = key;
            }
        }
        public static void main(String[] args) {
            // Init SparkSession
            SparkSession spark = SparkSession
                    .builder()
                    .master("yarn")
                    .appName("spark-hbase-tutorial")
                    .getOrCreate();
    
            // Data Schema
            String catalog = "{"+"\"table\":{\"namespace\":\"default\", \"name\":\"my_table\"}," +
                    "\"rowkey\":\"key\"," +
                    "\"columns\":{" +
                    "\"key\":{\"cf\":\"rowkey\", \"col\":\"key\", \"type\":\"string\"}," +
                    "\"name\":{\"cf\":\"cf\", \"col\":\"name\", \"type\":\"string\"}" +
                    "}" +
                    "}";
    
            Map<String, String> optionsMap = new HashMap<String, String>();
            optionsMap.put(HBaseTableCatalog.tableCatalog(), catalog);
    
            Dataset<Row> ds= spark.createDataFrame(Arrays.asList(
                    new SampleData("key1", "foo"),
                    new SampleData("key2", "bar")), SampleData.class);
    
            // Write to HBase
            ds.write()
                    .format("org.apache.hadoop.hbase.spark")
                    .options(optionsMap)
                    .option("hbase.spark.use.hbasecontext", "false")
                    .mode("overwrite")
                    .save();
    
            // Read from HBase
            Dataset dataset = spark.read()
                    .format("org.apache.hadoop.hbase.spark")
                    .options(optionsMap)
                    .option("hbase.spark.use.hbasecontext", "false")
                    .load();
            dataset.show();
        }
    }
    

    Python

    from pyspark.sql import SparkSession
    
    # Initialize Spark Session
    spark = SparkSession \
      .builder \
      .master('yarn') \
      .appName('spark-hbase-tutorial') \
      .getOrCreate()
    
    data_source_format = ''
    
    # Create some test data
    df = spark.createDataFrame(
        [
            ("key1", "foo"),
            ("key2", "bar"),
        ],
        ["key", "name"]
    )
    
    # Define the schema for catalog
    catalog = ''.join("""{
        "table":{"namespace":"default", "name":"my_table"},
        "rowkey":"key",
        "columns":{
            "key":{"cf":"rowkey", "col":"key", "type":"string"},
            "name":{"cf":"cf", "col":"name", "type":"string"}
        }
    }""".split())
    
    # Write to HBase
    df.write.format('org.apache.hadoop.hbase.spark').options(catalog=catalog).option("hbase.spark.use.hbasecontext", "false").mode("overwrite").save()
    
    # Read from HBase
    result = spark.read.format('org.apache.hadoop.hbase.spark').options(catalog=catalog).option("hbase.spark.use.hbasecontext", "false").load()
    result.show()

    Menjalankan kode

    1. Buka terminal sesi Cloud Shell.

    2. Buat clone repositori GitHub GoogleCloudDataproc/cloud-dataproc ke terminal sesi Cloud Shell Anda:

      git clone https://github.com/GoogleCloudDataproc/cloud-dataproc.git
      

    3. Ubah ke direktori cloud-dataproc/spark-hbase:

      cd cloud-dataproc/spark-hbase
      
      Contoh output:
      user-name@cloudshell:~/cloud-dataproc/spark-hbase (project-id)$
      

    4. Kirimkan tugas Dataproc.

    Java

    1. Tetapkan versi komponen dalam file pom.xml.
      1. Halaman versi rilis 2.0.x Dataproc mencantumkan versi komponen Scala, Spark, dan HBase yang diinstal dengan empat versi subminor 2.0 image terakhir dan terbaru.
        1. Untuk menemukan versi subminor cluster versi image 2.0, klik nama cluster di halaman Clusters di konsolGoogle Cloud untuk membuka halaman Cluster details, tempat Image version cluster dicantumkan.
      2. Atau, Anda dapat menjalankan perintah berikut di terminal sesi SSH dari node master cluster untuk menentukan versi komponen:
        1. Periksa versi scala:
          scala -version
          
        2. Periksa versi Spark (control-D untuk keluar):
          spark-shell
          
        3. Periksa versi HBase:
          hbase version
          
        4. Identifikasi dependensi versi Spark, Scala, dan HBase di Maven pom.xml:
          <properties>
            <scala.version>scala full version (for example, 2.12.14)</scala.version>
            <scala.main.version>scala main version (for example, 2.12)</scala.main.version>
            <spark.version>spark version (for example, 3.1.2)</spark.version>
            <hbase.client.version>hbase version (for example, 2.2.7)</hbase.client.version>
            <hbase-spark.version>1.0.0(the current Apache HBase Spark Connector version)>
          </properties>
          
          Catatan: hbase-spark.version adalah versi konektor Spark HBase saat ini; biarkan nomor versi ini tidak berubah.
      3. Edit file pom.xml di editor Cloud Shell untuk memasukkan nomor versi Scala, Spark, dan HBase yang benar. Klik Open Terminal setelah Anda selesai mengedit untuk kembali ke command line terminal Cloud Shell.
        cloudshell edit .
        
      4. Beralih ke Java 8 di Cloud Shell. Versi JDK ini diperlukan untuk membangun kode (Anda dapat mengabaikan pesan peringatan plugin):
        sudo update-java-alternatives -s java-1.8.0-openjdk-amd64 && export JAVA_HOME=/usr/lib/jvm/java-8-openjdk-amd64
        
      5. Verifikasi penginstalan Java 8:
        java -version
        
        Contoh output:
        openjdk version "1.8..."
         
    2. Buat file jar:
      mvn clean package
      
      File .jar ditempatkan di subdirektori /target (misalnya, target/spark-hbase-1.0-SNAPSHOT.jar.
    3. Kirim tugas.

      gcloud dataproc jobs submit spark \
          --class=hbase.SparkHBaseMain  \
          --jars=target/filename.jar \
          --region=cluster-region \
          --cluster=cluster-name
      
      • --jars: Masukkan nama file .jar Anda setelah "target/" dan sebelum ".jar".
      • Jika Anda tidak menetapkan classpath HBase driver dan executor Spark saat membuat cluster, Anda harus menetapkannya dengan setiap pengiriman tugas dengan menyertakan flag ‑‑properties berikut dalam perintah pengiriman tugas:
        --properties='spark.driver.extraClassPath=/etc/hbase/conf:/usr/lib/hbase/*,spark.executor.extraClassPath=/etc/hbase/conf:/usr/lib/hbase/*'
               

    4. Lihat output tabel HBase di output terminal sesi Cloud Shell:

      Waiting for job output...
      ...
      +----+----+
      | key|name|
      +----+----+
      |key1| foo|
      |key2| bar|
      +----+----+
      

    Python

    1. Kirim tugas.

      gcloud dataproc jobs submit pyspark scripts/pyspark-hbase.py \
          --region=cluster-region \
          --cluster=cluster-name
      
      • Jika Anda tidak menetapkan classpath HBase driver dan executor Spark saat membuat cluster, Anda harus menetapkannya dengan setiap pengiriman tugas dengan menyertakan flag ‑‑properties berikut dalam perintah pengiriman tugas:
        --properties='spark.driver.extraClassPath=/etc/hbase/conf:/usr/lib/hbase/*,spark.executor.extraClassPath=/etc/hbase/conf:/usr/lib/hbase/*'
               

    2. Lihat output tabel HBase di output terminal sesi Cloud Shell:

      Waiting for job output...
      ...
      +----+----+
      | key|name|
      +----+----+
      |key1| foo|
      |key2| bar|
      +----+----+
      

    Memindai tabel HBase

    Anda dapat memindai konten tabel HBase dengan menjalankan perintah berikut di terminal sesi SSH node master yang Anda buka di Verifikasi penginstalan konektor:

    1. Buka shell HBase:
      hbase shell
      
    2. Pindai 'my-table':
      scan 'my_table'
      
      Contoh output:
      ROW               COLUMN+CELL
       key1             column=cf:name, timestamp=1647364013561, value=foo
       key2             column=cf:name, timestamp=1647364012817, value=bar
      2 row(s)
      Took 0.5009 seconds
      

      Pembersihan

      Setelah menyelesaikan tutorial, Anda dapat membersihkan resource yang dibuat agar resource tersebut berhenti menggunakan kuota dan dikenai biaya. Bagian berikut menjelaskan cara menghapus atau menonaktifkan resource ini.

      Menghapus project

      Cara termudah untuk menghilangkan penagihan adalah dengan menghapus project yang Anda buat untuk tutorial.

      Untuk menghapus project:

      1. In the Google Cloud console, go to the Manage resources page.

        Go to Manage resources

      2. In the project list, select the project that you want to delete, and then click Delete.
      3. In the dialog, type the project ID, and then click Shut down to delete the project.

      Menghapus cluster

      • Untuk menghapus cluster Anda:
        gcloud dataproc clusters delete cluster-name \
            --region=${REGION}