परिचय
ईटीएल/ईएलटी पाइपलाइनों से लेकर मशीन लर्निंग मॉडल बनाने तक सभी प्रकार की समस्याओं को हल करने के लिए डेटा इंजीनियरों और डेटा वैज्ञानिकों के बीच पायथन का व्यापक रूप से उपयोग किया जाता है। Apache HBase कई वर्कफ़्लोज़ के लिए एक प्रभावी डेटा स्टोरेज सिस्टम है, लेकिन इस डेटा को विशेष रूप से पायथन के माध्यम से एक्सेस करना एक संघर्ष हो सकता है। डेटा पेशेवरों के लिए जो HBase में संग्रहीत डेटा का उपयोग करना चाहते हैं, हाल के अपस्ट्रीम प्रोजेक्ट "hbase-connectors" का उपयोग बुनियादी कार्यों के लिए PySpark के साथ किया जा सकता है।
इस ब्लॉग श्रृंखला में, हम बताएंगे कि कैसे बुनियादी स्पार्क उपयोग के साथ-साथ CDSW में बनाए गए कार्यों के लिए PySpark और HBase को एक साथ कॉन्फ़िगर किया जाए। जो लोग CDSW से परिचित नहीं हैं, उनके लिए यह डेटा वैज्ञानिकों के लिए अपनी स्वयं की एनालिटिक्स पाइपलाइनों का प्रबंधन करने के लिए एक सुरक्षित, स्वयं-सेवा उद्यम डेटा विज्ञान मंच है, इस प्रकार मशीन सीखने की परियोजनाओं को अन्वेषण से उत्पादन तक तेज करता है। CDSW के बारे में अधिक जानकारी के लिए Cloudera Data Science Workbench उत्पाद पृष्ठ पर जाएँ।
इस पोस्ट में, उदाहरण आउटपुट के साथ कई ऑपरेशनों को समझाया और प्रदर्शित किया जाएगा। संदर्भ के लिए, इस विशिष्ट ब्लॉग पोस्ट में सभी उदाहरण संचालन CDSW परिनियोजन के साथ चलाए जाते हैं।
आवश्यकताएं:
- एचबेस और स्पार्क के साथ एक सीडीपी क्लस्टर रखें
- यदि आप CDSW के माध्यम से उदाहरणों का अनुसरण करने जा रहे हैं, तो आपको इसे स्थापित करने की आवश्यकता होगी - क्लाउडेरा डेटा साइंस वर्कबेंच स्थापित करना
- पायथन 3 प्रत्येक नोड पर एक ही पथ पर स्थापित है
कॉन्फ़िगरेशन:
सबसे पहले, स्पार्क SQL क्वेरी को ठीक से काम करने के लिए HBase और Spark को एक साथ कॉन्फ़िगर करने की आवश्यकता है। ऐसा करने के लिए इसके दो भाग हैं:पहला, Cloudera प्रबंधक के माध्यम से HBase क्षेत्र सर्वर को कॉन्फ़िगर करें; और दूसरा, सुनिश्चित करें कि स्पार्क रन-टाइम में HBase बाइंडिंग है। हालाँकि, ध्यान में रखने वाली एक बात यह है कि Cloudera प्रबंधक पहले से ही आपके लिए HBase पर स्पार्क को स्वचालित रूप से इंगित करने के लिए कुछ कॉन्फ़िगरेशन और पर्यावरण चर सेट करता है। फिर भी, स्पार्क एसक्यूएल प्रश्नों को कॉन्फ़िगर करने का पहला चरण सीडीपी क्लस्टर पर सभी प्रकार के परिनियोजन के माध्यम से सामान्य है, लेकिन दूसरा परिनियोजन के प्रकार के आधार पर थोड़ा अलग है।
HBase क्षेत्र सर्वर को कॉन्फ़िगर करना
- क्लौडेरा मैनेजर पर जाएं और HBase सर्विस को चुनें।
- “regionserver Environment” खोजें
- RegionServer Environment Advanced Configuration Snippet (सुरक्षा वाल्व) का उपयोग करके एक नया पर्यावरण चर जोड़ें:
- कुंजी:HBASE_CLASSPATH
- मान:/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
सुनिश्चित करें कि आप उपयुक्त संस्करण संख्याओं का उपयोग करते हैं।
- क्षेत्र सर्वर पुनरारंभ करें।
एक बार जब आप ऊपर दिए गए चरणों का पालन कर लेते हैं, तो आप CDSW या गैर-CDSW परिनियोजन चाहते हैं, इसके आधार पर नीचे दिए गए चरणों का पालन करें।
गैर-CDSW परिनियोजन में स्पार्क रनटाइम में HBase बाइंडिंग जोड़ना
शेल को तैनात करने या स्पार्क-सबमिट का सही तरीके से उपयोग करने के लिए, स्पार्क में सही HBase बाइंडिंग सुनिश्चित करने के लिए निम्न कमांड का उपयोग करें।
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. जार
स्पार्क-सबमिट-जार/ऑप्ट/क्लाउडेरा/पार्सल/सीडीएच/लिब/एचबेस_कनेक्टर्स/लिब/एचबेस-स्पार्क.जर,/ऑप्ट/क्लाउडेरा/पार्सल/सीडीएच/लिब/एचबेस_कनेक्टर्स/लिब/एचबेस-स्पार्क-प्रोटोकॉल- छायांकित जार
CDSW परिनियोजन में स्पार्क रनटाइम में HBase बाइंडिंग जोड़ना
CDSW को HBase और PySpark के साथ कॉन्फ़िगर करने के लिए, आपको कुछ कदम उठाने होंगे।
1) सुनिश्चित करें कि प्रत्येक क्लस्टर नोड पर पायथन 3 स्थापित है और इसके पथ को नोट करें
2) CDSW में एक नया प्रोजेक्ट बनाएं और PySpark टेम्प्लेट का उपयोग करें
3) प्रोजेक्ट खोलें, सेटिंग्स -> इंजन -> पर्यावरण चर पर जाएं।
4) सेट करें PYSPARK3_DRIVER_PYTHON और PYSPARK3_PYTHON उस पथ पर जहां आपके क्लस्टर नोड्स पर पायथन स्थापित है (चरण 1 में नोट किया गया पथ)।
नीचे एक नमूना है कि इसे कैसा दिखना चाहिए।
5) अपने प्रोजेक्ट में, फाइल्स -> स्पार्क-डिफॉल्ट्स.कॉन्फ पर जाएं और इसे वर्कबेंच में खोलें
6) उस फाइल में नीचे की लाइन को कॉपी और पेस्ट करें और सुनिश्चित करें कि नया सत्र शुरू करने से पहले इसे सहेजा गया है।
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
इस बिंदु पर, CDSW अब HBase पर PySpark जॉब चलाने के लिए कॉन्फ़िगर किया गया है! इस ब्लॉग पोस्ट का शेष भाग CDSW परिनियोजन पर कुछ नमूना संचालनों को संदर्भित करता है।
उदाहरण संचालन
ऑपरेशन लगाएं
HBase में पंक्तियों को सम्मिलित और अद्यतन करने के दो तरीके हैं। पहला और सबसे अनुशंसित तरीका एक कैटलॉग बनाना है, जो एक स्कीमा है जो टेबल-नाम और नेमस्पेस को निर्दिष्ट करते हुए एक HBase टेबल के कॉलम को एक PySpark डेटाफ्रेम में मैप करेगा। इस उपयोगकर्ता-परिभाषित JSON प्रारूप का निर्माण सबसे पसंदीदा तरीका है क्योंकि इसका उपयोग अन्य कार्यों के साथ भी किया जा सकता है। कैटलॉग के बारे में अधिक जानकारी के लिए, यह दस्तावेज़ http://hbase.apache.org/book.html#_define_catalog देखें। दूसरी विधि "hbase.columns.mapping" नामक एक विशिष्ट मैपिंग पैरामीटर का उपयोग कर रही है जो केवल कुंजी-मूल्य जोड़े की एक स्ट्रिंग लेता है।
- कैटलॉग का उपयोग करना
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 सत्यापित करें कि HBase शेल को खोलकर और निम्नलिखित कमांड को निष्पादित करके "tblEmployee" नामक एक नई तालिका HBase में बनाई गई है:
स्कैन 'tblEmployee', {'LIMIT' => 2}
कैटलॉग का उपयोग करने से आप HBase तालिकाओं को आसानी से लोड कर सकते हैं। इस पर भविष्य की किस्त में चर्चा की जाएगी।
- hbase.columns.mapping का उपयोग करना
PySpark Dataframe लिखते समय, "hbase.columns.mapping" नामक एक विकल्प को एक स्ट्रिंग शामिल करने के लिए जोड़ा जा सकता है जो कॉलम को सही ढंग से मैप करता है। यह विकल्प आपको केवल मौजूदा तालिकाओं में पंक्तियों को सम्मिलित करने की अनुमति देता है।
HBase शेल में, आइए पहले एक टेबल बनाएं 'tblEmployee2', 'personal'
बनाएं
अब PySpark में "hbase.columns.mapping"
. का उपयोग करके 2 पंक्तियाँ डालें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() दोबारा, बस सत्यापित करें कि "tblEmployee2" नामक एक नई तालिका में ये नई पंक्तियां हैं।
स्कैन 'tblEmployee2', {'LIMIT' => 2}
यह हमारे उदाहरणों को पूरा करता है कि पाइस्पार्क के माध्यम से HBase तालिकाओं में पंक्तियों को कैसे सम्मिलित किया जाए। अगली किस्त में, मैं गेट एंड स्कैन ऑपरेशंस, पायस्पार्क एसक्यूएल और कुछ समस्या निवारण पर चर्चा करूंगा। तब तक आपको एक सीडीपी क्लस्टर प्राप्त करना चाहिए और इन उदाहरणों के माध्यम से अपना काम करना चाहिए।
