Pengantar
Python digunakan secara luas di antara Insinyur Data dan Ilmuwan Data untuk menyelesaikan segala macam masalah mulai dari pipeline ETL/ELT hingga membangun model pembelajaran mesin. Apache HBase adalah sistem penyimpanan data yang efektif untuk banyak alur kerja, tetapi mengakses data ini secara khusus melalui Python bisa menjadi perjuangan. Untuk profesional data yang ingin menggunakan data yang disimpan di HBase, proyek upstream terbaru “hbase-connectors” dapat digunakan dengan PySpark untuk operasi dasar.
Dalam seri blog ini, kami akan menjelaskan cara mengonfigurasi PySpark dan HBase bersama-sama untuk penggunaan Spark dasar serta untuk pekerjaan yang dipertahankan di CDSW. Bagi mereka yang tidak terbiasa dengan CDSW, ini adalah platform ilmu data perusahaan swalayan yang aman bagi ilmuwan data untuk mengelola saluran analitik mereka sendiri, sehingga mempercepat proyek pembelajaran mesin dari eksplorasi hingga produksi. Untuk informasi lebih lanjut tentang CDSW, kunjungi halaman produk Cloudera Data Science Workbench.
Dalam posting ini, beberapa operasi akan dijelaskan dan didemonstrasikan bersama dengan contoh output. Untuk konteks, semua operasi contoh dalam entri blog khusus ini dijalankan dengan penerapan CDSW.
Prasyarat:
- Memiliki cluster CDP dengan HBase dan Spark
- Jika Anda akan mengikuti contoh melalui CDSW, Anda perlu menginstalnya – Menginstal Cloudera Data Science Workbench
- Python 3 diinstal pada setiap node di jalur yang sama
Konfigurasi:
Pertama, HBase dan Spark perlu dikonfigurasi bersama agar kueri Spark SQL berfungsi dengan baik. Untuk melakukannya, ada dua bagian:pertama, konfigurasikan Server Wilayah HBase melalui Cloudera Manager; dan kedua, pastikan run-time Spark memiliki binding HBase. Namun, satu catatan yang perlu diingat adalah Cloudera Manager telah menyiapkan beberapa konfigurasi dan variabel lingkungan untuk secara otomatis mengarahkan Spark ke HBase untuk Anda. Namun demikian, langkah pertama dalam mengonfigurasi kueri Spark SQL adalah umum melalui semua jenis penerapan pada kluster CDP, tetapi langkah kedua sedikit berbeda tergantung pada jenis penerapan.
Mengonfigurasi Server Wilayah HBase
- Buka Cloudera Manager dan pilih layanan HBase.
- Telusuri “regionserver environment”
- Tambahkan variabel lingkungan baru menggunakan Cuplikan Konfigurasi Lanjut Lingkungan RegionServer (Katup Pengaman):
- Kunci:HBASE_CLASSPATH
- Nilai:/opt/cloudera/parcels/CDH/lib/hbase_connectors/lib/hbase-spark.jar:/opt/cloudera/parcels/CDH/lib/hbase_connectors/lib /hbase-spark-protocol-shaded.jar:/opt/cloudera/parcels/CDH/jars/scala-library-2.11.12.jar
Pastikan Anda menggunakan nomor versi yang sesuai.
- Mulai Ulang Server Wilayah.
Setelah Anda mengikuti langkah-langkah di atas, ikuti langkah-langkah di bawah ini tergantung apakah Anda menginginkan penerapan CDSW atau Non-CDSW.
Menambahkan Binding HBase ke Spark Runtime di Deployment Non-CDSW
Untuk men-deploy shell atau menggunakan spark-submit dengan benar, gunakan perintah berikut untuk memastikan spark memiliki binding HBase yang tepat.
pyspark –jars /opt/cloudera/parcels/CDH/lib/hbase_connectors/lib/hbase-spark.jar,/opt/cloudera/parcels/CDH/lib/hbase_connectors/lib/hbase-spark-protocol-shaded. toples
spark-submit –jars /opt/cloudera/parcels/CDH/lib/hbase_connectors/lib/hbase-spark.jar,/opt/cloudera/parcels/CDH/lib/hbase_connectors/lib/hbase-spark-protocol- shaded.jar
Menambahkan HBase Bindings ke Spark Runtime di Deployment CDSW
Untuk mengkonfigurasi CDSW dengan HBase dan PySpark, ada beberapa langkah yang perlu Anda ambil.
1) Pastikan Python 3 diinstal pada setiap node cluster dan catat jalurnya
2) Buat Proyek baru di CDSW dan gunakan template PySpark
3) Buka Proyek, buka Pengaturan -> Mesin -> Variabel Lingkungan.
4) Setel PYSPARK3_DRIVER_PYTHON dan PYSPARK3_PYTHON ke jalur tempat Python diinstal pada node cluster Anda (Jalur dicatat pada Langkah 1).
Di bawah ini adalah contoh bagaimana seharusnya terlihat.
5) Di Proyek Anda, buka Files -> spark-defaults.conf dan buka di Workbench
6) Salin dan tempel baris di bawah ini di file itu dan pastikan itu disimpan sebelum memulai sesi baru.
spark.jars=/opt/cloudera/parcels/CDH/lib/hbase_connectors/lib/hbase-spark.jar,/opt/cloudera/parcels/CDH/lib/hbase_connectors/lib/hbase-spark-protocol-shaded.jar
Pada titik ini, CDSW sekarang dikonfigurasi untuk menjalankan pekerjaan PySpark di HBase! Sisa dari entri blog ini mengacu pada beberapa contoh operasi pada penerapan CDSW.
Contoh Operasi
Operasi Pasang
Ada dua cara untuk menyisipkan dan memperbarui baris ke HBase. Metode pertama dan paling direkomendasikan adalah membuat katalog, yang merupakan skema yang akan memetakan kolom tabel HBase ke kerangka data PySpark sambil menentukan nama tabel dan namespace. Membangun format JSON yang ditentukan pengguna ini adalah metode yang paling disukai karena dapat digunakan dengan operasi lain juga. Untuk informasi lebih lanjut tentang katalog, lihat dokumentasi ini http://hbase.Apache.org/book.html#_define_catalog. Metode kedua menggunakan parameter pemetaan khusus yang disebut “hbase.columns.mapping” yang hanya mengambil string pasangan nilai kunci.
- Menggunakan Katalog
from pyspark.sql import Row
from pyspark.sql import SparkSession
spark = SparkSession\
.builder\
.appName("SampleApplication")\
.getOrCreate()
tableCatalog = ''.join("""{
"table":{"namespace":"default", "name":"tblEmployee", "tableCoder":"PrimitiveType"},
"rowkey":"key",
"columns":{
"key":{"cf":"rowkey", "col":"key", "type":"int"},
"empId":{"cf":"personal","col":"empId","type":"string"},
"empName":{"cf":"personal", "col":"empName", "type":"string"},
"empState":{"cf":"personal", "col":"empWeight", "type":"string"}
}
}""".split())
employee = [(10, 'jonD', 'Jon Daniels', 'CA'), (6, 'billR', 'Bill Robert', 'FL')]
employeeRDD = spark.sparkContext.parallelize(employee)
employeeMap = employeeRDD.map(lambda x: Row(key=int(x[0]), empId=x[1], empName=x[2], empState=x[3]))
employeeDF = spark.createDataFrame(employeeMap)
employeeDF.write.format("org.apache.hadoop.hbase.spark") \
.options(catalog=tableCatalog, newTable=5) \
.option("hbase.spark.use.hbasecontext", False) \
.save()
# newTable refers to the NumberOfRegions which has to be > 3 Verifikasi bahwa tabel baru bernama “tblEmployee” dibuat di HBase hanya dengan membuka shell HBase dan menjalankan perintah berikut:
pindai ‘tblEmployee’, {‘LIMIT’ => 2}
Menggunakan katalog juga memungkinkan Anda memuat tabel HBase dengan mudah. Ini akan dibahas dalam angsuran mendatang.
- Menggunakan hbase.columns.mapping
Saat menulis PySpark Dataframe, opsi yang disebut "hbase.columns.mapping" dapat ditambahkan untuk menyertakan string yang memetakan kolom dengan benar. Opsi ini hanya memungkinkan Anda untuk menyisipkan baris ke tabel yang ada.
Di shell HBase, mari kita buat tabel dulu buat 'tblEmployee2', 'personal'
Sekarang di PySpark mari kita masukkan 2 baris menggunakan “hbase.columns.mapping”
from pyspark.sql import Row
from pyspark.sql import SparkSession
spark = SparkSession\
.builder\
.appName("SampleApplication")\
.getOrCreate()
employee = [(10, 'jonD', 'Jon Daniels', 170.7), (6, 'billR', 'Bill Robert', 200.1)]
employeeRDD = spark.sparkContext.parallelize(employee)
employeeMap = employeeRDD.map(lambda x: Row(key=int(x[0]), empId=x[1], empName=x[2], empWeight=float(x[3])))
employeeDF = spark.createDataFrame(employeeMap)
employeeDF.write.format("org.apache.hadoop.hbase.spark") \
.option("hbase.columns.mapping", "key INTEGER :key, empId STRING personal:empId, empName STRING personal:empName, empWeight FLOAT personal:empWeight") \
.option("hbase.table", "tblEmployee2") \
.option("hbase.spark.use.hbasecontext", False) \
.save() Sekali lagi, cukup verifikasi bahwa tabel baru bernama “tblEmployee2” memiliki baris baru ini.
pindai ‘tblEmployee2’, {‘LIMIT’ => 2}
Itu melengkapi contoh kami tentang cara menyisipkan baris melalui PySpark ke tabel HBase. Dalam angsuran berikutnya, saya akan membahas Get and Scan Operations, PySpark SQL, dan beberapa pemecahan masalah. Sampai saat itu, Anda harus mendapatkan cluster CDP dan mengerjakan contoh-contoh berikut.
