Giới thiệu
Python được sử dụng rộng rãi giữa các Kỹ sư dữ liệu và Nhà khoa học dữ liệu để giải quyết tất cả các loại vấn đề từ đường ống ETL / ELT đến xây dựng mô hình học máy. Apache HBase là một hệ thống lưu trữ dữ liệu hiệu quả cho nhiều quy trình làm việc nhưng việc truy cập dữ liệu này cụ thể thông qua Python có thể là một cuộc đấu tranh. Đối với các chuyên gia dữ liệu muốn sử dụng dữ liệu được lưu trữ trong HBase, dự án ngược dòng gần đây “hbase-connector” có thể được sử dụng với PySpark cho các hoạt động cơ bản.
Trong loạt bài blog này, chúng tôi sẽ giải thích cách cấu hình PySpark và HBase cùng nhau để sử dụng Spark cơ bản cũng như cho các công việc được duy trì trong CDSW. Đối với những người không quen thuộc với CDSW, đây là một nền tảng khoa học dữ liệu doanh nghiệp tự phục vụ, an toàn cho các nhà khoa học dữ liệu quản lý các đường ống phân tích của riêng họ, do đó đẩy nhanh các dự án máy học từ thăm dò đến sản xuất. Để biết thêm thông tin về CDSW, hãy truy cập trang sản phẩm Cloudera Data Science Workbench.
Trong bài đăng này, một số thao tác sẽ được giải thích và minh họa cùng với kết quả ví dụ. Đối với ngữ cảnh, tất cả các thao tác ví dụ trong bài đăng blog cụ thể này được chạy với triển khai CDSW.
Điều kiện tiên quyết:
- Có một cụm CDP với HBase và Spark
- Nếu bạn định làm theo các ví dụ qua CDSW, bạn sẽ cần cài đặt nó - Cài đặt Cloudera Data Science Workbench
- Python 3 được cài đặt trên mỗi nút tại cùng một đường dẫn
Cấu hình:
Đầu tiên, HBase và Spark cần được cấu hình cùng nhau để các truy vấn Spark SQL hoạt động bình thường. Để làm như vậy, có hai phần:đầu tiên, định cấu hình Máy chủ Vùng HBase thông qua Trình quản lý Cloudera; và thứ hai, đảm bảo thời gian chạy Spark có các ràng buộc HBase. Tuy nhiên, một lưu ý cần ghi nhớ là Trình quản lý Cloudera đã thiết lập một số biến cấu hình và môi trường để tự động trỏ Spark tại HBase cho bạn. Tuy nhiên, bước đầu tiên của việc định cấu hình các truy vấn Spark SQL là phổ biến thông qua tất cả các kiểu triển khai trên các cụm CDP, nhưng bước thứ hai hơi khác một chút tùy thuộc vào kiểu triển khai.
Định cấu hình Máy chủ Vùng HBase
- Đi tới Trình quản lý Cloudera và chọn dịch vụ HBase.
- Tìm kiếm “môi trường máy chủ vùng”
- Thêm một biến môi trường mới bằng cách sử dụng Đoạn trích cấu hình nâng cao của môi trường RegionServer (Van an toàn):
- Khóa:HBASE_CLASSPATH
- Giá trị:/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
Đảm bảo bạn sử dụng số phiên bản thích hợp.
- Khởi động lại Máy chủ Vùng.
Khi bạn làm theo các bước trên, hãy làm theo các bước bên dưới tùy thuộc vào việc bạn muốn triển khai CDSW hay Non-CDSW.
Thêm liên kết HBase vào Spark Runtime trong các triển khai không phải CDSW
Để triển khai shell hoặc sử dụng spark-submit một cách chính xác, hãy sử dụng các lệnh sau để đảm bảo spark có các ràng buộc HBase phù hợp.
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. lọ
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
Thêm liên kết HBase vào Spark Runtime trong các triển khai CDSW
Để cấu hình CDSW với HBase và PySpark, bạn cần thực hiện một số bước.
1) Đảm bảo Python 3 được cài đặt trên mỗi nút cụm và lưu ý đường dẫn đến nó
2) Tạo một Dự án mới trong CDSW và sử dụng mẫu PySpark
3) Mở Dự án, đi tới Cài đặt -> Động cơ -> Biến môi trường.
4) Đặt PYSPARK3_DRIVER_PYTHON và PYSPARK3_PYTHON đến đường dẫn nơi Python được cài đặt trên các nút cụm của bạn (Đường dẫn được lưu ý trong Bước 1).
Dưới đây là một mẫu về cách nó sẽ trông như thế nào.
5) Trong Dự án của bạn, đi tới Tệp -> spark-defaults.conf và mở nó trong Workbench
6) Sao chép và dán dòng bên dưới vào tệp đó và đảm bảo rằng nó đã được lưu trước khi bắt đầu một phiên mới.
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
Tại thời điểm này, CDSW hiện đã được cấu hình để chạy các công việc PySpark trên HBase! Phần còn lại của bài đăng trên blog này đề cập đến một số thao tác mẫu khi triển khai CDSW.
Hoạt động Ví dụ
Đặt Hoạt động
Có hai cách để chèn và cập nhật hàng vào HBase. Phương pháp đầu tiên và được đề xuất nhiều nhất là xây dựng một danh mục, là một lược đồ sẽ ánh xạ các cột của bảng HBase với khung dữ liệu PySpark trong khi chỉ định tên bảng và không gian tên. Xây dựng định dạng JSON do người dùng xác định này là phương pháp được ưa thích nhất vì nó cũng có thể được sử dụng với các hoạt động khác. Để biết thêm thông tin về danh mục, hãy tham khảo tài liệu này http://hbase.apache.org/book.html#_define_catalog. Phương pháp thứ hai là sử dụng một tham số ánh xạ cụ thể được gọi là “hbase.columns.mapping”, tham số này chỉ nhận một chuỗi các cặp khóa-giá trị.
- Sử dụng Catalogs
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 Xác minh rằng một bảng mới có tên “tblErantyee” được tạo trong HBase bằng cách chỉ cần mở HBase shell và thực hiện lệnh sau:
quét ‘tblErantyee’, {‘LIMIT’ => 2}
Sử dụng danh mục cũng có thể cho phép bạn tải bảng HBase dễ dàng. Điều này sẽ được thảo luận trong phần sau.
- Sử dụng hbase.columns.mapping
Trong khi viết Khung dữ liệu PySpark, một tùy chọn có tên “hbase.columns.mapping” có thể được thêm vào để bao gồm một chuỗi ánh xạ các cột một cách chính xác. Tùy chọn này chỉ cho phép bạn chèn các hàng vào bảng hiện có.
Trong HBase shell, trước tiên chúng ta hãy tạo một bảng tạo ‘tblEFastee2’, ‘personal’
Bây giờ trong PySpark, hãy chèn 2 hàng bằng cách sử dụng “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() Một lần nữa, chỉ cần xác minh rằng bảng mới có tên “tblEFastee2” có các hàng mới này.
quét ‘tblErantyee2’, {‘LIMIT’ => 2}
Điều đó hoàn thành các ví dụ của chúng tôi về cách chèn các hàng thông qua PySpark vào bảng HBase. Trong phần tiếp theo, tôi sẽ thảo luận về Thao tác lấy và quét, PySpark SQL và một số cách khắc phục sự cố. Cho đến lúc đó, bạn sẽ có được một cụm CDP và làm việc theo cách của bạn thông qua các ví dụ này.
