अपाचे काफ्का और स्पार्क स्ट्रीमिंग के साथ स्ट्रीमिंग

नमस्कार, हेब्र! आज हम एक प्रणाली का निर्माण करेंगे जो अपार्क काफ्का का उपयोग स्पार्क स्ट्रीमिंग का उपयोग करके संदेश धाराओं को संसाधित करने और एडब्ल्यूएस आरडीएस क्लाउड डेटाबेस में प्रसंस्करण परिणाम लिखने के लिए करेगी।

कल्पना कीजिए कि एक निश्चित क्रेडिट संस्थान ने हमें अपनी सभी शाखाओं में मक्खी पर आने वाले लेनदेन को संसाधित करने का कार्य निर्धारित किया है। यह ट्रेजरी के लिए खुली मुद्रा स्थिति, लेनदेन पर सीमा या वित्तीय परिणाम आदि की जल्दी से गणना करने के लिए किया जा सकता है।

जादू और जादू मंत्र के उपयोग के बिना इस मामले को कैसे लागू किया जाए - हम कटौती के तहत पढ़ते हैं! चलो चलते हैं!


(छवि स्रोत)

परिचय

बेशक, वास्तविक समय में एक बड़े डेटा सरणी को संसाधित करना आधुनिक प्रणालियों में उपयोग के लिए पर्याप्त अवसर प्रदान करता है। इसके लिए सबसे लोकप्रिय संयोजनों में से एक है अपाचे काफ्का और स्पार्क स्ट्रीमिंग टैंडम, जहां काफ्का आने वाले संदेश पैकेटों की एक धारा बनाता है, और स्पार्क स्ट्रीमिंग इन पैकेटों को एक निर्दिष्ट समय अंतराल पर संसाधित करता है।

आवेदन की गलती सहिष्णुता को बढ़ाने के लिए, हम चौकियों - चौकियों का उपयोग करेंगे। इस तंत्र का उपयोग करते हुए, जब स्पार्क स्ट्रीमिंग मॉड्यूल को खोए हुए डेटा को पुनर्प्राप्त करने की आवश्यकता होती है, तो इसे केवल अंतिम नियंत्रण बिंदु पर लौटने और इसके साथ गणना फिर से शुरू करने की आवश्यकता होती है।

विकास के तहत प्रणाली का आर्किटेक्चर

प्रयुक्त घटक:

• अपाचे काफ्का प्रकाशित और सदस्यता के साथ एक वितरित संदेश प्रणाली है। ऑफ़लाइन और ऑनलाइन संदेश खपत दोनों के लिए उपयुक्त है। डेटा हानि को रोकने के लिए, काफ्का संदेशों को डिस्क पर संग्रहीत किया जाता है और क्लस्टर के भीतर दोहराया जाता है। काफ्का प्रणाली को चिड़ियाघर कीपर सिंक्रोनाइज़ेशन सेवा के शीर्ष पर बनाया गया है;
• अपाचे स्पार्क स्ट्रीमिंग - स्ट्रीमिंग डेटा को संसाधित करने के लिए स्पार्क घटक। स्पार्क स्ट्रीमिंग मॉड्यूल को माइक्रो-बैच आर्किटेक्चर का उपयोग करके बनाया गया है, जब डेटा स्ट्रीम को छोटे डेटा पैकेट के निरंतर अनुक्रम के रूप में व्याख्या की जाती है। स्पार्क स्ट्रीमिंग विभिन्न स्रोतों से डेटा प्राप्त करती है और उन्हें छोटे पैकेटों में जोड़ती है। नियमित अंतराल पर नए पैकेज बनाए जाते हैं। प्रत्येक समय अंतराल की शुरुआत में, एक नया पैकेट बनाया जाता है, और इस अंतराल के दौरान प्राप्त किसी भी डेटा को पैकेट में शामिल किया जाता है। अंतराल के अंत में, पैकेट वृद्धि बंद हो जाती है। अंतराल का आकार एक पैरामीटर द्वारा निर्धारित किया जाता है जिसे बैच अंतराल कहा जाता है;
• अपाचे स्पार्क एसक्यूएल - स्पार्क कार्यात्मक प्रोग्रामिंग के साथ संबंधपरक प्रसंस्करण को जोड़ती है। संरचित डेटा उस डेटा को संदर्भित करता है जिसमें एक स्कीमा है, अर्थात् सभी रिकॉर्ड के लिए फ़ील्ड का एक सेट। स्पार्क SQL विभिन्न संरचित डेटा स्रोतों से इनपुट का समर्थन करता है और, स्कीमा के बारे में जानकारी की उपलब्धता के लिए धन्यवाद, यह केवल आवश्यक रिकॉर्ड फ़ील्ड को कुशलता से प्राप्त कर सकता है, और डेटाफ़्रेम एपीआई भी प्रदान करता है;
• AWS RDS एक अपेक्षाकृत सस्ता क्लाउड-आधारित रिलेशनल डेटाबेस है, जो एक वेब सेवा है जो कॉन्फ़िगरेशन, संचालन और स्केलिंग को सरल करता है, और सीधे अमेज़न द्वारा प्रशासित किया जाता है।

Kafka सर्वर को स्थापित और शुरू करें

सीधे काफ्का का उपयोग करने से पहले, आपको यह सुनिश्चित करने की आवश्यकता है कि जावा उपलब्ध है, जैसा कि जेवीएम का उपयोग कार्य के लिए किया जाता है:

sudo apt-get update sudo apt-get install default-jre java -version 

काफ्का के साथ काम करने के लिए एक नया उपयोगकर्ता बनाएँ:

 sudo useradd kafka -m sudo passwd kafka sudo adduser kafka sudo 

इसके बाद, आधिकारिक Apache Kafka वेबसाइट से वितरण डाउनलोड करें:

 wget -P /YOUR_PATH "http://apache-mirror.rbc.ru/pub/apache/kafka/2.2.0/kafka_2.12-2.2.0.tgz" 

डाउनलोड किए गए संग्रह को अनपैक करें:

 tar -xvzf /YOUR_PATH/kafka_2.12-2.2.0.tgz ln -s /YOUR_PATH/kafka_2.12-2.2.0 kafka 

अगला चरण वैकल्पिक है। तथ्य यह है कि डिफ़ॉल्ट सेटिंग्स Apache Kafka की सभी विशेषताओं का पूर्ण उपयोग करने की अनुमति नहीं देती हैं। उदाहरण के लिए, किसी विषय, श्रेणी, समूह को हटाएं, जिसमें संदेश प्रकाशित किए जा सकते हैं। इसे बदलने के लिए, कॉन्फ़िगरेशन फ़ाइल को संपादित करें:

 vim ~/kafka/config/server.properties 

फ़ाइल के अंत में निम्नलिखित जोड़ें:

 delete.topic.enable = true 

काफ्का सर्वर शुरू करने से पहले, आपको ज़ूकीपर सर्वर शुरू करने की आवश्यकता है, हम उस सहायक स्क्रिप्ट का उपयोग करेंगे जो काकाका वितरण के साथ आती है:

 Cd ~/kafka bin/zookeeper-server-start.sh config/zookeeper.properties 

ज़ूकिपर के सफलतापूर्वक शुरू होने के बाद, एक अलग टर्मिनल में हमने काफ्का सर्वर लॉन्च किया:

 bin/kafka-server-start.sh config/server.properties 

लेनदेन नामक एक नया विषय बनाएँ:

 bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 3 --topic transaction 

सुनिश्चित करें कि विभाजन और प्रतिकृति की सही संख्या वाला विषय बनाया गया है:

 bin/kafka-topics.sh --describe --zookeeper localhost:2181 

हम निर्माता और उपभोक्ता के नव निर्मित विषय के परीक्षण के क्षणों को याद करेंगे। संदेश भेजने और प्राप्त करने का परीक्षण करने के तरीके के बारे में अधिक जानकारी के लिए, आधिकारिक दस्तावेज़ देखें - कुछ संदेश भेजें । खैर, हम काफ्काप्रोड्यूसर एपीआई का उपयोग करके पायथन में एक निर्माता को लिखने के लिए आगे बढ़ते हैं।

निर्माता लेखन

निर्माता यादृच्छिक डेटा उत्पन्न करेगा - हर सेकंड 100 संदेश। यादृच्छिक डेटा से हमारा मतलब है एक शब्दकोश जिसमें तीन क्षेत्र शामिल हैं:

• शाखा - क्रेडिट संस्थान की बिक्री के बिंदु का नाम;
• मुद्रा - लेनदेन मुद्रा;
• राशि - लेनदेन राशि। यदि यह बैंक द्वारा मुद्रा की खरीद है, और बिक्री होने पर ऋणात्मक है तो यह राशि एक सकारात्मक संख्या होगी।

निर्माता के लिए कोड इस प्रकार है:

 from numpy.random import choice, randint def get_random_value(): new_dict = {} branch_list = ['Kazan', 'SPB', 'Novosibirsk', 'Surgut'] currency_list = ['RUB', 'USD', 'EUR', 'GBP'] new_dict['branch'] = choice(branch_list) new_dict['currency'] = choice(currency_list) new_dict['amount'] = randint(-100, 100) return new_dict 

अगला, भेजने की विधि का उपयोग करते हुए, हम सर्वर को एक संदेश भेजते हैं, जिस विषय में हमें जरूरत है, JSON प्रारूप में:

 from kafka import KafkaProducer producer = KafkaProducer(bootstrap_servers=['localhost:9092'], value_serializer=lambda x:dumps(x).encode('utf-8'), compression_type='gzip') my_topic = 'transaction' data = get_random_value() try: future = producer.send(topic = my_topic, value = data) record_metadata = future.get(timeout=10) print('--> The message has been sent to a topic: \ {}, partition: {}, offset: {}' \ .format(record_metadata.topic, record_metadata.partition, record_metadata.offset )) except Exception as e: print('--> It seems an Error occurred: {}'.format(e)) finally: producer.flush() 

स्क्रिप्ट चलाते समय, हमें टर्मिनल में निम्नलिखित संदेश मिलते हैं:


इसका मतलब यह है कि सब कुछ वैसा ही काम करता है जैसा हम चाहते थे - निर्माता उस विषय पर संदेश भेजता है और हमें आवश्यकता होती है।

अगला कदम स्पार्क को स्थापित करना और इस संदेश प्रवाह को संसाधित करना है।

अपाचे स्पार्क स्थापित करें

अपाचे स्पार्क एक बहुमुखी और उच्च प्रदर्शन क्लस्टर कंप्यूटिंग प्लेटफॉर्म है।

प्रदर्शन के संदर्भ में, स्पार्क मैपएड्यूस मॉडल के लोकप्रिय कार्यान्वयन को पार करता है, साथ ही साथ इंटरैक्टिव प्रश्नों और धारा प्रसंस्करण सहित गणना की एक विस्तृत श्रृंखला के लिए समर्थन प्रदान करता है। बड़ी मात्रा में डेटा को संसाधित करने में गति एक महत्वपूर्ण भूमिका निभाती है, क्योंकि यह गति है जो आपको मिनटों या घंटों के इंतजार के बिना अंतःक्रियात्मक रूप से काम करने की अनुमति देती है। इतनी तेज गति से स्पार्क की सबसे बड़ी ताकत इसकी मेमोरी-इन गणना करने की क्षमता है।

यह रूपरेखा स्काला में लिखी गई है, इसलिए आपको इसे पहले स्थापित करना होगा:

 sudo apt-get install scala 

आधिकारिक वेबसाइट से स्पार्क वितरण डाउनलोड करें:

 wget "http://mirror.linux-ia64.org/apache/spark/spark-2.4.2/spark-2.4.2-bin-hadoop2.7.tgz" 

संग्रह अनपैक करें:

 sudo tar xvf spark-2.4.2/spark-2.4.2-bin-hadoop2.7.tgz -C /usr/local/spark 

बैश फ़ाइल में स्पार्क का पथ जोड़ें:

 vim ~/.bashrc 

संपादक के माध्यम से निम्नलिखित पंक्तियाँ जोड़ें:

 SPARK_HOME=/usr/local/spark export PATH=$SPARK_HOME/bin:$PATH 

Bashrc में परिवर्तन करने के बाद नीचे दी गई कमांड चलाएँ:

 source ~/.bashrc 

AWS PostgreSQL परिनियोजन

यह डेटाबेस को तैनात करने के लिए बना हुआ है, जहां हम धाराओं से संसाधित जानकारी अपलोड करेंगे। इसके लिए हम AWS RDS सेवा का उपयोग करेंगे।

कंसोल पर जाएं AWS -> AWS RDS -> डेटाबेस -> डेटाबेस बनाएँ:


PostgreSQL का चयन करें और अगला बटन क्लिक करें:


क्योंकि यह उदाहरण पूरी तरह से शैक्षिक उद्देश्यों के लिए समझा जाता है, हम एक मुफ्त सर्वर "न्यूनतम" (फ्री टियर) का उपयोग करेंगे:


इसके बाद, फ्री टियर ब्लॉक में एक टिक लगाएं, और उसके बाद हम स्वचालित रूप से t2.micro क्लास का एक उदाहरण पेश करेंगे - हालाँकि यह कमजोर है, यह मुफ़्त है और हमारे कार्य के लिए काफी उपयुक्त है:

बहुत महत्वपूर्ण चीजें अनुसरण करती हैं: डेटाबेस उदाहरण का नाम, मास्टर उपयोगकर्ता का नाम और उसका पासवर्ड। आइए नाम का उदाहरण दें: myHabrTest, मास्टर उपयोगकर्ता: habr , पासवर्ड: habr12345 और नेक्स्ट पेज पर क्लिक करें:



अगले पृष्ठ में बाहर (सार्वजनिक पहुंच) और पोर्ट उपलब्धता से हमारे डेटाबेस सर्वर की उपलब्धता के लिए जिम्मेदार पैरामीटर हैं:


चलो VPC सुरक्षा समूह के लिए एक नया कॉन्फ़िगरेशन बनाते हैं, जो हमें पोर्ट 5432 (PostgreSQL) के माध्यम से बाहर से हमारे डेटाबेस सर्वर तक पहुंचने की अनुमति देगा।

एक अलग ब्राउज़र विंडो में, VPC डैशबोर्ड में AWS कंसोल पर जाएँ -> सुरक्षा समूह -> सुरक्षा समूह अनुभाग बनाएँ:

सुरक्षा समूह के लिए नाम सेट करें - PostgreSQL, एक विवरण, इंगित करें कि कौन सा VPC इस समूह के साथ जुड़ा होना चाहिए और बनाएँ बटन क्लिक करें:


पोर्ट 5432 के लिए नए बनाए गए इनबाउंड नियम समूह को भरें, जैसा कि नीचे दी गई तस्वीर में दिखाया गया है। आपको मैन्युअल पोर्ट निर्दिष्ट करने की आवश्यकता नहीं है, लेकिन टाइप ड्रॉप-डाउन सूची से PostgreSQL का चयन करें।

कड़ाई से बोलने पर, मूल्य :: / 0 का अर्थ है दुनिया भर के एक सर्वर के लिए आने वाले ट्रैफ़िक की उपलब्धता, जो कि कैनोनिक रूप से पूरी तरह से सच नहीं है, लेकिन उदाहरण को पार्स करने के लिए, आइए हम इस दृष्टिकोण का उपयोग करें:


हम उस ब्राउज़र पृष्ठ पर लौटते हैं, जहाँ हमारे पास "उन्नत सेटिंग्स कॉन्फ़िगर करें" खुली हुई है और अनुभाग VPC सुरक्षा समूहों में चुनें -> मौजूदा VPC सुरक्षा समूह चुनें -> PostgreSQL:


इसके बाद, अनुभाग में डेटाबेस विकल्प -> डेटाबेस नाम -> नाम सेट करें - habrDB ।

हम डिफ़ॉल्ट रूप से बैकअप (बैकअप प्रतिधारण अवधि - 0 दिन) को अक्षम करने, निगरानी और प्रदर्शन अंतर्दृष्टि के अपवाद के साथ, बाकी मापदंडों को छोड़ सकते हैं। डेटाबेस बनाएँ बटन पर क्लिक करें:

स्ट्रीम हैंडलर

अंतिम चरण स्पार्क-नौकरियों का विकास होगा, जो कफका से प्राप्त हर दो सेकंड के नए डेटा को संसाधित करेगा और परिणाम को डेटाबेस में दर्ज करेगा।

जैसा कि ऊपर उल्लेख किया गया है, चौकियों स्पार्कस्ट्रीमिंग में प्राथमिक तंत्र हैं जिन्हें गलती सहिष्णुता प्रदान करने के लिए कॉन्फ़िगर किया जाना चाहिए। हम नियंत्रण बिंदु का उपयोग करेंगे और, प्रक्रिया ड्रॉप होने की स्थिति में, स्पार्क स्ट्रीमिंग मॉड्यूल को केवल अंतिम नियंत्रण बिंदु पर लौटने और खोए हुए डेटा को पुनर्स्थापित करने के लिए इससे गणना फिर से शुरू करने की आवश्यकता होगी।

आप निर्देशिका को एक दोष-सहिष्णु, विश्वसनीय फ़ाइल सिस्टम (उदाहरण के लिए, एचडीएफएस, एस 3, आदि) में सेट करके चेकपॉइंट को सक्षम कर सकते हैं, जिसमें चेकपॉइंट की जानकारी सहेजी जाएगी। यह प्रयोग किया जाता है, उदाहरण के लिए:

 streamingContext.checkpoint(checkpointDirectory) 

हमारे उदाहरण में, हम निम्नलिखित दृष्टिकोण का उपयोग करेंगे, अर्थात्, यदि चौकीडायरेक्टरी मौजूद है, तो संदर्भ नियंत्रण डेटा डेटा से फिर से बनाया जाएगा। यदि निर्देशिका मौजूद नहीं है (यानी यह पहली बार निष्पादित किया गया है), functionToCreateContext फ़ंक्शन को एक नया संदर्भ बनाने और DStreams को कॉन्फ़िगर करने के लिए कहा जाता है:

 from pyspark.streaming import StreamingContext context = StreamingContext.getOrCreate(checkpointDirectory, functionToCreateContext) 

KafkaUtils लाइब्रेरी की CreateDirectStream विधि का उपयोग करके "लेन-देन" विषय से कनेक्ट करने के लिए एक DirectStream ऑब्जेक्ट बनाएँ:

 from pyspark.streaming.kafka import KafkaUtils sc = SparkContext(conf=conf) ssc = StreamingContext(sc, 2) broker_list = 'localhost:9092' topic = 'transaction' directKafkaStream = KafkaUtils.createDirectStream(ssc, [topic], {"metadata.broker.list": broker_list}) 

JSON प्रारूप में आने वाले डेटा को पार्स करना:

 rowRdd = rdd.map(lambda w: Row(branch=w['branch'], currency=w['currency'], amount=w['amount'])) testDataFrame = spark.createDataFrame(rowRdd) testDataFrame.createOrReplaceTempView("treasury_stream") 

स्पार्क एसक्यूएल का उपयोग करके, हम एक सरल समूह बनाते हैं और परिणाम को कंसोल पर प्रिंट करते हैं:

 select from_unixtime(unix_timestamp()) as curr_time, t.branch as branch_name, t.currency as currency_code, sum(amount) as batch_value from treasury_stream t group by t.branch, t.currency 

क्वेरी टेक्स्ट प्राप्त करना और इसे स्पार्क SQL के माध्यम से चलाना:

 sql_query = get_sql_query() testResultDataFrame = spark.sql(sql_query) testResultDataFrame.show(n=5) 

और फिर हम प्राप्त एकत्रित डेटा को AWS RDS में एक तालिका में सहेजते हैं। एकत्रीकरण के परिणामों को डेटाबेस तालिका में सहेजने के लिए, हम DataFrame ऑब्जेक्ट के लिखने की विधि का उपयोग करेंगे:

 testResultDataFrame.write \ .format("jdbc") \ .mode("append") \ .option("driver", 'org.postgresql.Driver') \ .option("url","jdbc:postgresql://myhabrtest.ciny8bykwxeg.us-east-1.rds.amazonaws.com:5432/habrDB") \ .option("dbtable", "transaction_flow") \ .option("user", "habr") \ .option("password", "habr12345") \ .save() 

AWS RDS से संबंध स्थापित करने के बारे में कुछ शब्द। हमने "AWS PostgreSQL" कदम पर इसके लिए उपयोगकर्ता और पासवर्ड बनाया। डेटाबेस सर्वर url के लिए, एंडपॉइंट का उपयोग करें, जो कनेक्टिविटी और सुरक्षा अनुभाग में प्रदर्शित होता है:

स्पार्क और काफ्का को सही ढंग से जोड़ने के लिए, आपको स्पार्क-स्ट्रीमिंग-काफ्का-0-8_2.11 विरूपण साक्ष्य का उपयोग करके स्मार्क-सबमिट के माध्यम से काम चलाना चाहिए। इसके अलावा, हम PostgreSQL डेटाबेस के साथ बातचीत करने के लिए कलाकृतियों को भी लागू करते हैं, हम उन्हें -पैकेज के माध्यम से स्थानांतरित करेंगे।

स्क्रिप्ट के लचीलेपन के लिए, हम संदेश सर्वर और उस विषय का नाम भी लेते हैं, जिससे हम इनपुट मापदंडों के रूप में डेटा प्राप्त करना चाहते हैं।

इसलिए, सिस्टम शुरू करने और परीक्षण करने का समय आ गया है:

 spark-submit \ --packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.0.2,\ org.postgresql:postgresql:9.4.1207 \ spark_job.py localhost:9092 transaction 

सब कुछ काम कर गया! जैसा कि आप नीचे दिए गए चित्र में देख सकते हैं, एप्लिकेशन के कार्य के दौरान, हर 2 सेकंड में नए एकत्रीकरण परिणाम प्रदर्शित होते हैं, क्योंकि हम बैचिंगिंग अंतराल को 2 सेकंड में सेट करते हैं जब StreamingContext ऑब्जेक्ट बनाते हैं:


इसके बाद, हम लेन-देन_प्रवाह तालिका में रिकॉर्ड की जांच करने के लिए डेटाबेस के लिए एक सरल प्रश्न बनाते हैं:

निष्कर्ष

इस लेख ने अपाचे कफ़्का और पोस्टग्रेक्यूएल के साथ संयोजन में स्पार्क स्ट्रीमिंग का उपयोग करके सूचना प्रसंस्करण स्ट्रीमिंग का एक उदाहरण देखा। विभिन्न स्रोतों से डेटा की वृद्धि के साथ, स्ट्रीमिंग अनुप्रयोगों और वास्तविक समय में परिचालन अनुप्रयोगों को बनाने के लिए स्पार्क स्ट्रीमिंग के व्यावहारिक मूल्य को कम करना मुश्किल है।

आप GitHub पर मेरे भंडार में पूर्ण स्रोत कोड पा सकते हैं।

मैं इस लेख पर खुशी के साथ चर्चा करने के लिए तैयार हूं, मैं आपकी टिप्पणियों का इंतजार कर रहा हूं, और मैं सभी संबंधित पाठकों की रचनात्मक आलोचना की भी उम्मीद करता हूं।

मैं आपको सफलता की कामना करता हूं!

PS मूल रूप से एक स्थानीय PostgreSQL डेटाबेस का उपयोग करने की योजना बनाई गई थी, लेकिन AWS के मेरे प्यार को देखते हुए, मैंने डेटाबेस को क्लाउड में रखने का निर्णय लिया। इस विषय पर अगले लेख में, मैं यह दिखाऊंगा कि AWS Kinesis और AWS KR का उपयोग करके AWS में ऊपर वर्णित पूरी प्रणाली को कैसे लागू किया जाए। खबर का पालन करें!