मनमानी मशीन लर्निंग मॉडल का उपयोग करके अपाचे स्पार्क पर बड़ी मात्रा में डेटा को वर्गीकृत करें

भाग 2: समाधान

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

प्रतिक्रियाशील धाराएँ संकल्पना

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

सामान्य तौर पर, रिएक्टिव स्ट्रीम प्रकाशक-सब्सक्राइबर टेम्पलेट को लागू करने के लिए एक विनिर्देश है। यह विनिर्देश चार इंटरफेस (प्रकाशक, सब्सक्राइबर, प्रोसेसर और सब्सक्रिप्शन) और उनके तरीकों के लिए एक अनुबंध को परिभाषित करता है।

आइए इन इंटरफेसों पर अधिक विस्तार से विचार करें:

public interface Publisher<T> { public void subscribe(Subscriber<? super T> s); } public interface Subscriber<T> { public void onSubscribe(Subscription s); public void onNext(T t); public void onError(Throwable t); public void onComplete(); } public interface Subscription { public void request(long n); public void cancel(); } public interface Processor<T, R> extends Subscriber<T>, Publisher<R> { } 

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

1. onSubscribe। इस विधि को सब्सक्राइबर को प्रकाशक के साथ पंजीकृत करने के तुरंत बाद कहा जाता है। एक पैरामीटर के रूप में, एक सदस्यता वस्तु इसके लिए पारित की जाती है जिसके माध्यम से सब्सक्राइबर प्रकाशक से डेटा का अनुरोध करेगा। इस ऑब्जेक्ट को केवल इस सबस्क्राइबर के संदर्भ में संग्रहीत किया जाना चाहिए।
2. सब्सक्राइबर द्वारा संबंधित सब्सक्रिप्शन ऑब्जेक्ट पर अनुरोध विधि को कॉल करके प्रकाशक से डेटा का अनुरोध करने के बाद, प्रकाशक अगले तत्व को पास करते हुए, सब्सक्राइबर को ऑन लाइन विधि से कॉल कर सकता है।
3. इसके बाद सब्सक्राइबर समय-समय पर सब्सक्रिप्शन पर रिक्वेस्ट मेथड को कॉल कर सकता है, लेकिन पब्लिशर रिक्वेस्ट मेथड के जरिए रिक्वेस्ट किए गए टोटल से ज्यादा नहीं कर सकता।
4. यदि डेटा स्ट्रीम परिमित है, तो सभी तत्वों को onNext विधि से पारित करने के बाद, प्रकाशक को onComplete विधि को कॉल करना होगा।
5. यदि प्रकाशक में कोई त्रुटि हुई है और तत्वों की आगे की प्रक्रिया संभव नहीं है, तो इसे ऑनरर विधि को कॉल करना चाहिए
6. ऑनक्रीम या ऑनरॉइड विधियों को कॉल करने के बाद, सब्सक्राइबर के साथ प्रकाशक की आगे की बातचीत को बाहर रखा जाना चाहिए।

विधि कॉल को प्रकाशक और सब्सक्राइबर के बीच सिग्नल भेजने के रूप में माना जा सकता है। सब्सक्राइबर ने प्रकाशक को संकेत दिया कि यह कितने तत्वों को संसाधित करने के लिए तैयार है, और प्रकाशक, बदले में, यह संकेत देता है कि या तो अगला तत्व है, या कोई और तत्व नहीं हैं, या कुछ त्रुटि हुई है।

एक दूसरे पर प्रकाशक और सब्सक्राइबर के दूसरे प्रभाव को बाहर करने के लिए, प्रतिक्रियाशील धाराओं के इंटरफेस को लागू करने वाले सभी तरीकों को कॉल करना गैर-अवरुद्ध होना चाहिए। इस मामले में, उनके बीच की बातचीत पूरी तरह से अतुल्यकालिक होगी।

प्रतिक्रियाशील धाराओं के इंटरफेस के विनिर्देश के बारे में अधिक विवरण यहां पाया जा सकता है ।

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

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

एक चौकस पाठक तुरंत सवाल पूछेगा: अवरुद्ध कतार क्यों है, क्योंकि प्रतिक्रियाशील धाराओं विनिर्देश के अनुसार, सभी तरीकों का कार्यान्वयन गैर-अवरुद्ध होना चाहिए? लेकिन यह सब यहीं है: चूंकि हम तत्वों की कड़ाई से परिभाषित संख्या के लिए प्रकाशक से पूछ रहे हैं, इसलिए ऑननेक्स्ट विधि को इस संख्या से अधिक नहीं कहा जाएगा, और कतार हमेशा अवरुद्ध किए बिना एक नया तत्व जोड़ सकती है।

दूसरी ओर, जब एक खाली कतार के मामले में हैनक्स्ट विधि कहा जाता है तो ब्लॉकिंग हो सकती है। हालाँकि, यह सब ठीक है: हैनक्स्ट विधि सब्सक्राइबर इंटरफ़ेस के अनुबंध का हिस्सा नहीं है, यह Iterator इंटरफ़ेस में परिभाषित किया गया है, जिसे हमने पहले बताया था, एक अवरुद्ध डेटा संरचना है। अगली विधि को कॉल करते समय, हम अगले तत्व को कतार से घटाते हैं, और जब इसका आकार एक निश्चित सीमा से कम हो जाता है, तो हमें कॉल के माध्यम से तत्वों के अगले हिस्से को अनुरोध विधि से अनुरोध करना होगा।

चित्रा 7. प्रतिक्रियाशील धाराओं दृष्टिकोण का उपयोग करके एक बाहरी सेवा के साथ अतुल्यकालिक बातचीत

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

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

अक्का-धाराएँ और एक्का-http प्रतिक्रियाशील धाराओं के विनिर्देश के कार्यान्वयन के रूप में

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

अक्का में अभिनेताओं की अवधारणा के कार्यान्वयन के बारे में बहुत सारे साहित्य लिखे गए हैं, इसलिए हम यहां नहीं रुकते हैं ( अक्का आधिकारिक साइट जानकारी का एक बहुत अच्छा स्रोत है, मैं एक्शन में भी अक्का की सिफारिश करता हूं)। यहां हम JVM के तहत कार्यान्वयन के तकनीकी पक्ष पर करीब से नज़र डालेंगे।

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

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

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

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

आइए चरणों के स्रोत, प्रवाह और सिंक पर अधिक विस्तार से विचार करें। स्रोत डेटा स्रोत को परिभाषित करता है। अक्करा-धाराएँ एक दर्जन से अधिक विभिन्न तरीकों का समर्थन करती हैं, जो एक इट्रेटर सहित स्रोत बनाने के लिए हैं:

 val featuresSource: Source[Array[Float], NotUsed] = Source.fromIterator { () => featuresIterator } 

मौजूदा स्रोत को परिवर्तित करके भी स्रोत प्राप्त किया जा सकता है:

 val newSource: Source[String, NotUsed] = source.map(item => transform(item)) 

यदि परिवर्तन एक nontrivial ऑपरेशन है, तो इसे एक प्रवाह इकाई के रूप में दर्शाया जा सकता है। अक्का-धारा प्रवाह बनाने के लिए कई अलग-अलग तरीकों का समर्थन करती है। सबसे आसान तरीका एक फ़ंक्शन से बनाना है:

 val someFlow: Flow[String, Int, NotUsed] = Flow.fromFunction((x: String) => x.length) 

सोर्स एंड फ्लो को मिलाकर हमें एक नया सोर्स मिलता है।

 val newSource: Source[Int, NotUsed] = oldSource.via(someFlow) 

सिंक को डाटा प्रोसेसिंग के अंतिम चरण के रूप में उपयोग किया जाता है। जैसा कि स्रोत के मामले में, अक्का-धाराएं एक दर्जन से अधिक विभिन्न सिंक विकल्प प्रदान करती हैं, उदाहरण के लिए, Sink.foreach प्रत्येक तत्व के लिए एक निश्चित ऑपरेशन करता है, सिंक एक संग्रह में सभी तत्वों को इकट्ठा करता है, आदि।

 val printSink: Sink[Any, Future[Done]] = Sink.foreach(println) 

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

आवश्यक स्रोत, प्रवाह, और सिंक को मिलाकर, हमें RunnableGraph मिलता है। यह एक प्रकार से परिचालित होता है, जो इस ग्राफ के निष्पादन के परिणामस्वरूप प्राप्त मूल्य के प्रकार को निर्धारित करता है। यदि आवश्यक हो, तो चरणों का संयोजन करते समय, आप निर्दिष्ट कर सकते हैं कि चरणों में से कौन सा चरणों के संचालन के पूरे ग्राफ का परिणाम होगा। डिफ़ॉल्ट रूप से, स्रोत चरण का परिणाम लिया जाता है:

 val graph: RunnableGraph[NotUsed] = someSource.to(Sink.foreach(println)) 

हालांकि, अगर सिंक चरण का परिणाम हमारे लिए अधिक महत्वपूर्ण है, तो हमें स्पष्ट रूप से यह संकेत देना चाहिए:

 val graph: RunnableGraph[Future[Done]] = someSource.toMat(Sink.foreach(println))(Keep.right) 

ऑपरेशन के ग्राफ को परिभाषित करने के बाद, हमें इसे चलाना होगा। ऐसा करने के लिए, runnableGraph को रन विधि को कॉल करना होगा। एक पैरामीटर के रूप में, यह विधि एक ActorMaterializer ऑब्जेक्ट लेता है (जो एक अंतर्निहित दायरे में भी हो सकता है), जो कि उन अभिनेताओं को बनाने के लिए जिम्मेदार है जो ऑपरेशन करेंगे। आमतौर पर, एक ActorMaterializer एक ActorSystem के निर्माण के तुरंत बाद बनाया जाता है, अपने जीवन चक्र से जुड़ा होता है, और इसका उपयोग अभिनेताओं को बनाने के लिए करता है। एक उदाहरण पर विचार करें:

 //   ActorSystem,       implicit val system = ActorSystem(“system name”) // ,       ActorSystem implicit val materializer = ActorMaterializer() //    ,       Sink val graph: RunnableGraph[Future[immutable.Seq[Int]]] = Source.fromIterator(() => (1 to 10).iterator).toMat(Sink.seq)(Keep.right) //   ,    implicit scope. val result: Future[immutable.Seq[Int]] = graph.run() 

सरल संयोजनों के मामले में, आप एक अलग RunnableGraph बनाने के बिना कर सकते हैं, लेकिन बस स्रोत को सिंक से कनेक्ट करें और उन्हें स्रोत पर रनविथ विधि कहकर शुरू करें। यह विधि यह भी मानती है कि एक ActorMaterializer ऑब्जेक्ट निहित दायरे में मौजूद है। इसके अलावा, इस मामले में, सिंक में परिभाषित भौतिक मूल्य का उपयोग किया जाएगा। उदाहरण के लिए, निम्नलिखित कोड का उपयोग करके, हम स्रोत को प्रतिक्रियाशील धाराओं के विनिर्देश से प्रकाशक में बदल सकते हैं:

 val source: Source[Score, NotUsed] = Source.fromIterator(() => sourceIterator).map(item => transform(item)) val publisher: Publisher[Score] = source.runWith(Sink.asPublisher(false)) 

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

Akka की संरचना में akka -http मॉड्यूल शामिल है, जो आपको HTTP पर अतुल्यकालिक गैर-अवरुद्ध संचार को व्यवस्थित करने की अनुमति देता है। इसके अलावा, यह मॉड्यूल एक्का धाराओं के आधार पर बनाया गया है, जो आपको डेटा फ्लो प्रोसेसिंग ऑपरेशंस के ग्राफ में अतिरिक्त चरण के रूप में HTTP इंटरैक्शन को जोड़ने की अनुमति देता है।

बाहरी सेवाओं से जुड़ने के लिए, एक्का-http तीन अलग-अलग इंटरफेस प्रदान करता है।

1. अनुरोध-स्तर एपीआई - एक मनमानी मशीन के लिए एकल अनुरोधों के मामले में सबसे सरल विकल्प है। इस स्तर पर, HTTP कनेक्शन पूरी तरह से स्वचालित रूप से प्रबंधित किए जाते हैं, और प्रत्येक अनुरोध में मशीन का पूरा पता प्रेषित करना आवश्यक है जिसमें अनुरोध भेजा जा रहा है।
2. होस्ट-स्तरीय एपीआई - उपयुक्त जब हम जानते हैं कि हम किस मशीन पर पोर्ट का उपयोग करेंगे। इस स्थिति में, एक्का-HTTP HTTP कनेक्शन के पूल का नियंत्रण लेता है, और अनुरोधों में यह अनुरोधित संसाधन के सापेक्ष पथ को निर्दिष्ट करने के लिए पर्याप्त है।
3. कनेक्शन-स्तर एपीआई - आपको HTTP कनेक्शन के प्रबंधन पर पूर्ण नियंत्रण प्राप्त करने की अनुमति देता है, अर्थात्, कनेक्शन भर में अनुरोधों को खोलना, बंद करना और वितरित करना।

हमारे मामले में, वर्गीकरण सेवा का पता हमें पहले से ज्ञात है, इसलिए, केवल इस विशेष मशीन के साथ HTTP इंटरैक्शन को व्यवस्थित करना आवश्यक है। इसलिए, होस्ट-स्तरीय एपीआई हमारे लिए सबसे अच्छा है। अब, आइए देखें कि इसका उपयोग करते समय HTTP कनेक्शन का पूल कैसे बनाया जाता है:

 val httpFlow: Flow[(HttpRequest,Id), (Try[HttpResponse],Id), Http.HostConnectionPool] = Http().cachedHostConnectionPool[Id](hostAddress, portNumber) 

जब Http ()। CachedHostConnectionPool [T] (hostAddress, portNumber) को कॉल कर रहा है, तो ActorSystem में, जो कि एक निहित दायरे में है, संसाधनों को कनेक्शन पूल बनाने के लिए आवंटित किया जाता है, लेकिन कनेक्शन स्वयं स्थापित नहीं होते हैं। इस कॉल के परिणामस्वरूप, फ़्लो लौटाया जाता है, जो एक HTTP अनुरोध और इनपुट के रूप में कुछ आईडी पहचान ऑब्जेक्ट प्राप्त करता है। इस तथ्य के कारण संबंधित प्रतिक्रिया के साथ अनुरोध को मिलान करने के लिए पहचान ऑब्जेक्ट की आवश्यकता होती है, क्योंकि HTTP कॉल inkka-http एक अतुल्यकालिक ऑपरेशन है, और जिस क्रम में प्रतिक्रियाएं प्राप्त होती हैं, जरूरी नहीं कि जिस क्रम में अनुरोध भेजा गया है, उसके अनुरूप हो। इसलिए, आउटपुट फ्लो में क्वेरी के परिणाम और संबंधित पहचान ऑब्जेक्ट के कुछ जोड़े देता है।

सीधे तौर पर, HTTP कनेक्शन तब स्थापित होता है जब कोई ग्राफ़ (इस प्रवाह सहित) को लॉन्च किया जाता है (भौतिकीकृत)। अक्का-http को इस तरह से लागू किया जाता है कि चाहे कितनी भी बार httpFlow वाले ग्राफ़ को भौतिक रूप में बदल दिया गया हो, एक एक्टरसिस्टम के भीतर हमेशा HTTP कनेक्शन का एक सामान्य पूल होगा जो सभी भौतिकता द्वारा उपयोग किया जाएगा। यह आपको नेटवर्क संसाधनों के उपयोग को बेहतर ढंग से नियंत्रित करने और उन्हें अधिभार से बचने की अनुमति देता है।

इस प्रकार, HTTP कनेक्शन पूल का जीवन चक्र ActorSystem से जुड़ा हुआ है। जैसा कि पहले ही उल्लेख किया गया है, थ्रेड पूल का जीवन चक्र भी इसके साथ जुड़ा हुआ है, जिसमें अभिनेताओं (या हमारे मामले में, चरण akka-streams और akka-http) के रूप में परिभाषित संचालन किए जाते हैं। इसलिए, अधिकतम दक्षता प्राप्त करने के लिए, हमें उसी JVM प्रक्रिया के भीतर ActorSystem के एक उदाहरण का पुन: उपयोग करना होगा।

यह सब एक साथ रखना: वर्गीकरण सेवा के साथ बातचीत को लागू करने का एक उदाहरण

तो, अब हम बाहरी सेवाओं के साथ अतुल्यकालिक बातचीत का उपयोग करते हुए अपाचे स्पार्क पर वितरित डेटा के बड़े संस्करणों को वर्गीकृत करने की प्रक्रिया पर आगे बढ़ सकते हैं। इस इंटरैक्शन की सामान्य योजना पहले ही चित्र 7 में दिखाई जा चुकी है।

मान लीजिए कि हमारे पास कुछ प्रारंभिक डेटासेट [विशेषताएं] परिभाषित हैं। मैपपार्टिशन ऑपरेशन को उस पर लागू करते हुए, हमें एक डेटासैट प्राप्त करना चाहिए, जिसमें स्रोत सेट से प्रत्येक आईडी को वर्गीकरण (डेटासेट [स्कोर]) के परिणामस्वरूप प्राप्त एक निश्चित मूल्य के साथ मोहर दी गई है। निष्पादकों पर अतुल्यकालिक प्रसंस्करण को व्यवस्थित करने के लिए, हमें स्रोत और परिणामस्वरूप पुनरावृत्तियों को प्रकाशक और सब्सक्राइबर में क्रमशः प्रतिक्रियाशील धाराओं के विनिर्देश से लपेटना चाहिए और उन्हें एक साथ जोड़ना होगा।

 case class Features(id: String, vector: Array[Float]) case class Score(id: String, score: Float) //(1) val batchesRequestCount = config.getInt(“scoreService. batchesRequestCount”)//(2) //... val scoreDs: Dataset[Score] = featuresDs.mapPartitions { fi: Iterator[Features] => val publisher: Publisher[Iterable[Score]] = createPublisher(fi) //(3) val iteratorSubscriber: Iterator[Score] = new IteratorSubscriber(batchesRequestCount) //(4) publisher.subscribe(batchesRequestCount) //(5) iteratorSubscriber //(6) } 

इस कार्यान्वयन में, यह ध्यान में रखा जाता है कि एक कॉल के लिए वर्गीकरण सेवा एक बार फीचर वैक्टर के एक समूह को संसाधित कर सकती है, इसलिए, एक कॉल के बाद वर्गीकरण परिणाम भी पूरे समूह के लिए तुरंत उपलब्ध होगा। इसलिए, प्रकाशक के लिए एक पैरामीटर प्रकार के रूप में, हमारे पास केवल स्कोर नहीं है, जैसा कि आप उम्मीद कर सकते हैं, लेकिन Iterable [स्कोर]। इस प्रकार, हम इस समूह के लिए परिणामी इट्रेटर (जो कि एक सब्सक्राइबर भी है) के लिए वर्गीकरण परिणामों को ऑन कॉल विधि से सिंगल कॉल द्वारा भेजते हैं। यह प्रत्येक तत्व के लिए onNext को कॉल करने की तुलना में बहुत अधिक कुशल है। अब हम इस कोड का और अधिक विस्तार से विश्लेषण करेंगे।

1. हम इनपुट और आउटपुट डेटा की संरचना निर्धारित करते हैं। इनपुट के रूप में, हमारे पास एक सुविधा वेक्टर के साथ कुछ आईडी पहचानकर्ता का एक गुच्छा होगा, और आउटपुट के रूप में, हमारे पास वर्गीकरण के परिणामस्वरूप प्राप्त संख्यात्मक मान के साथ पहचानकर्ता का एक गुच्छा होगा।
2. हम एक बार में प्रकाशक से अनुरोध करने वाले समूहों की संख्या निर्धारित करते हैं। चूंकि यह माना जाता है कि ये मान बफर में झूठ होंगे और जब तक वे परिणामी पुनरावृत्ति से पढ़े जाते हैं, तब तक प्रतीक्षा करेंगे, यह मान निष्पादक को आवंटित स्मृति की मात्रा पर निर्भर करता है।
3. स्रोत इट्रेटर से प्रकाशक बनाएँ। वह वर्गीकरण सेवा के साथ बातचीत के लिए जिम्मेदार होगा। CreatePublisher फ़ंक्शन नीचे चर्चा की गई है।
4. एक सब्सक्राइबर बनाएं, जिसके परिणामस्वरूप इट्रेटर होगा। IteratorSubscriber वर्ग कोड भी नीचे दिया गया है।
5. प्रकाशक के साथ पंजीकरण करना।
6. मैपपार्टिशन ऑपरेशन के परिणाम के रूप में IteratorSubscriber लौटें।

अब createPublisher फ़ंक्शन के कार्यान्वयन पर विचार करें।

 type Ids = Seq[String] //(1) val batchSize = config.getInt("scoreService.batchSize") val parallelismLevel = config.getInt("scoreService.parallelismLevel") //(2) //... def createPublisher(fi: Iterator[Features]): Publisher[Iterable[Score]] = { import ActorSystemHolder._ //(3) Source .fromIterator(() => fi) //(4) .grouped(batchSize) //(5) .map { groupedFeatures: Seq[Features] => val request: (HttpRequest, Ids) = createHttpRequest(groupedFeatures) //(6) logger.debug(s"Sending request for the first id: ${request._2(0)}") request } .via(httpFlow) //(7) .flatMapMerge(parallelismLevel, { //(8) case (Success(response), ids) if response.status.isSuccess() => logger.debug(s"Processing successful result for the first id: ${ids(0)}") val resultSource: Source[Iterable[Score], _] = response.entity.dataBytes.reduce(_ ++ _).map { responseBytes => processSuccessfulResponse(responseBytes, ids) } //(9) resultScore case (Success(response), ids) => logger.warn( s"Failed result for the first id: ${ids(0)}, HTTP status: ${response.status}" ) response.discardEntityBytes() Source.failed( new IOException(s"Non-successful HTTP status: ${response.status}") ) //(10) case (Failure(ex), ids) => logger.warn(s"Failed result: an exception has occured", ex) Source.failed(ex) //(11) }) .runWith(Sink.asPublisher(false)) //(12) } def createHttpRequest(featuresSeq: Seq[Features]): (HttpRequest, ProfileIds) = { val requestBytes: Array[Byte] = featuresToMatrixBytes(featuresSeq) val ids: ProfileIds = extractIds(featuresSeq) val httpRequest = HttpRequest( method = HttpMethods.PUT, uri = "/score", entity = requestBytes ) httpRequest -> ids } 

1. - , . httpFlow, .
2. : , (batchSize) (parallelismLevel).
3. implicit scope ActorSystem, ActorMaterializer httpFlow. Spark-. ActorSystemHolder .
4. akka-streams . Source[Features] .
5. batchSize .
6. HttpRequest . HttpRequest createHttpRequest. createPublisher. feature-, , ( predict). , HTTP-. , HTTP-, HTTP-, URI .
7. httpFlow.
8. , . flatMapMerge, akka-http Source[ByteString], , . . parallelismLevel , ( ). HTTP-: , , , .
9. : . akka ByteString. , ByteString O(1), ByteString . , , . , .
10. HTTP- , Stream . , discardEntityBytes , , .
11. . akka-http , .
12. , Publisher, . , . false Sink.asPublisher , Publisher Subscriber-.

जैसा कि हमने पिछले खंड में उल्लेख किया है, अक्का के साथ काम करने के लिए आपको एक एक्टरसिस्ट की आवश्यकता है, जिसे एक बार बनाया जाना चाहिए और फिर पुन: उपयोग किया जाना चाहिए। दुर्भाग्य से, हमारे पास स्पार्क निष्पादक के वैश्विक वातावरण को कॉल करने का कोई तरीका नहीं है, लेकिन हम वैश्विक वस्तुओं को बनाने के लिए मानक तरीकों का सहारा ले सकते हैं। चूंकि स्पार्क निष्पादक एक अलग JVM प्रक्रिया है, इसलिए, इसके भीतर, हम एक वैश्विक ऑब्जेक्ट बना सकते हैं जिसमें हम ActorSystem और ActorMatrializer और httpFlow को संग्रहीत करेंगे।

 object ActorSystemHolder { implicit lazy val actorSystem: ActorSystem = { //(1) val actorSystemName = s"score-service-client" logger.debug(s"Creating actor system $actorSystemName") val as = ActorSystem(actorSystemName) //(2) logger.debug("Adding shutdown hook for the actor system") scala.sys.addShutdownHook { //(3) logger.debug(s"Terminating actor system $actorSystemName") Await.result(as.terminate(), 30.seconds) //to Mars :) logger.debug(s"The actor system $actorSystemName has been terminated") } as } implicit lazy val materializer: ActorMaterializer = { //(4) logger.debug(s"Creating actor materializer for actor system ${actorSystem.name}") ActorMaterializer() } lazy val httpFlow: Flow[ (HttpRequest,ProfileIds), (Try[HttpResponse], ProfileIds), Http.HostConnectionPool] = { //(5) val httpFlowSettings = ConnectionPoolSettings(actorSystem) logger.debug(s"Creating http flow with settings $httpFlowSettings") Http().cachedHostConnectionPool[ProfileIds]( config.getString("scoreService.host"), config.getInt("scoreService.int"), settings = httpFlowSettings ) } } 

1. हम आलसी इनिशियलाइज़ेशन का उपयोग करके सभी वैश्विक वैरिएबल्स बनाते हैं, अर्थात वास्तव में वे तब बनाए जाएंगे जब वे पहली बार माँग में होंगे।
2. यह एक विशिष्ट नाम के साथ एक नया ActorSystem बनाता है।
3. एक्टरसिस्टम के ढांचे के भीतर निष्पादित सभी प्रक्रियाओं को सही ढंग से समाप्त करने के लिए, हमें उस पर समाप्ति की विधि को कॉल करना होगा, जो बदले में, सभी अभिनेताओं को उनके मानक स्टॉप तंत्र का उपयोग करने से रोक देगा। ऐसा करने के लिए, हमें उस हुक को पंजीकृत करना होगा जिसे JVM प्रक्रिया समाप्त होने पर कहा जाता है।
4. अगला, हम एक ActorMaterializer बनाते हैं जो हमारे ActorSystem का उपयोग करके एक्का-स्ट्रीम प्रक्रियाओं का निष्पादन शुरू करेगा।
5. अंत में, हम बाहरी सेवा के साथ बातचीत करने के लिए httpFlow बनाते हैं। जैसा कि पिछले भाग में बताया गया है, यहाँ हम ActorSystem के ढांचे के भीतर HTTP कनेक्शनों के पूल के लिए संसाधन आवंटित करते हैं।

अब परिणामस्वरूप HTTP के कार्यान्वयन को हमारी HTTP इंटरैक्शन प्रक्रिया में एक सब्सक्राइबर के रूप में लागू करने पर विचार करें।

 sealed trait QueueItem[+T] case class Item[+T](item: T) extends QueueItem[T] case object Done extends QueueItem[Nothing] case class Failure(cause: Throwable) extends QueueItem[Nothing] //(1) class StreamErrorCompletionException(cause: Throwable) extends Exception(cause) //(2) class IteratorSubscriber[T](requestSize: Int) extends Subscriber[Iterable[T]] with Iterator[T] { //(3) private val buffer: BlockingQueue[QueueItem[Iterable[T]]] = new LinkedBlockingQueue[QueueItem[Iterable[T]]]() //(4) private var expecting: Int = 0 //(5) private val subscriptionPromise: Promise[Subscription] = Promise() private lazy val subscription: Subscription = Await.result(subscriptionPromise.future, 5.minutes) //(6) private var currentIterator: Iterator[T] = Iterator.empty //(7) private var isDone = false //(8) override def onSubscribe(s: Subscription): Unit = { subscriptionPromise.success(s) //(9) logger.trace("The iterator has been subscribed") } override def onNext(t: Iterable[T]): Unit = { logger.trace("Putting a next batch to the buffer") buffer.put(Item(t)) //(10) } override def onComplete(): Unit = { logger.debug("The stream has been succesfully completed") buffer.put(Done) //(11) } override def onError(t: Throwable): Unit = { logger.warn("The stream has been completed with error", t) buffer.put(Failure(t)) //(12) } override def hasNext: Boolean = { logger.trace("Asking hasNext") if (currentIterator.hasNext) { //(13) true } else if (isDone) { //(14) false } else { if (expecting < requestSize) { requestNextBatches() //(15) } buffer.take() match { //(16) case Item(batch) => currentIterator = batch.iterator expecting -= 1 this.hasNext //(17) case Done => isDone = true false //(18) case Failure(exception) => throw new StreamErrorCompletionException(exception) //(19) } } } override def next(): T = { val out = currentIterator.next() logger.trace("The next element is {}", out) out //(20) } private def requestNextBatches(): Unit = { logger.debug(s"Requesting {} batches", requestSize) subscription.request(requestSize) expecting += requestSize //(21) } } 

IteratorSubscriber वर्ग निर्माता-उपभोक्ता मॉडल का कार्यान्वयन है। सब्सक्राइबर इंटरफ़ेस को लागू करने वाला हिस्सा निर्माता है, और जो भाग Iterator को लागू करता है वह उपभोक्ता है। अवरुद्ध कतार के रूप में कार्यान्वित बफर का उपयोग संचार के साधन के रूप में किया जाता है। Iterator इंटरफ़ेस के तरीकों को Apache Spark निष्पादक पूल से एक धारा में कहा जाता है, और सब्सक्राइबर इंटरफ़ेस के तरीके ActorSystem के स्वामित्व वाले पूल से बुलाए जाते हैं।

अब, आइए दिए गए IteratorSubscriber कार्यान्वयन कोड को और अधिक विस्तार से देखें।

1. सबसे पहले, हम बफर तत्वों के लिए एक बीजीय डेटा प्रकार को परिभाषित करते हैं। बफर में, हमारे पास निम्नलिखित तत्वों का समूह हो सकता है, या डोन के सफल समापन का संकेत हो सकता है, या विफलता का संकेत हो सकता है, जिसमें थ्रोबेबल शामिल है, जो त्रुटि का कारण बना।
2. , hasNext .
3. , , Publisher-.
4. , . LinkedBlockingQueue, . , .
5. , . , , Publisher-. , , Publisher- . hasNext next ( requestNextBatches hasNext), , .
6. subscriptionPromise subscription Subscription, Publisher onSubscribe. , Reactive Streams Subscriber- Publisher- , , hasNext , onSubscribe. , subscription, Publisher-. lazy subscription, Promise.
7. . hasNext next, , .
8. , , hasNext false . hasNext, .
9. onSubscribe Publisher- Subscription Promise, subscription.
10. onNext Publisher-, . .
11. Publisher onComplete, Done.
12. Publisher onError. .
13. hasNext , . , true, . , .
14. , false.
15. , , requestSize, Publisher. , , , Publisher- , HTTP- .
16. . , , , . , , ( , , subscription), , , , .
17. , currentIterator. , . , hasNext , ( , ), .
18. , false hasNext. , isDone, , . - , hasNext , false. , hasNext , false , . , .
19. , , , .
20. अगली विधि वर्तमान पुनरावृत्ति से अगला तत्व लौटाता है। इसके कॉल के शब्दार्थ के अनुसार, इससे पहले, कॉल करने वाले को hasNext विधि को कॉल करना होगा, इसलिए अगले कॉल करते समय, अगला तत्व हमेशा अगला तत्व होना चाहिए।
21. यहां हम प्रकाशक को एक संकेत भेजते हैं कि हम प्रकाशक के साथ पंजीकरण करते समय प्राप्त सदस्यता वस्तु का उपयोग करके परिणामों के अगले समूह को संसाधित करने के लिए तैयार हैं। समूहों की संख्या अनुरोध के मूल्य से निर्धारित होती है। हम इस मान से अपेक्षित तत्वों की संख्या भी बढ़ाते हैं।

इस प्रकार, डेटा ब्लॉक के लिए पूर्ण प्रसंस्करण की सामान्य योजना, निष्पादक पर लॉन्च की गई, इस तरह दिखाई देती है:

चित्रा 8. स्रोत और परिणामस्वरूप पुनरावृत्तियों के साथ अभिनेताओं की सहभागिता।

निष्कर्ष में: इस समाधान के फायदे और नुकसान

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

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

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

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

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

नेटवर्क इंटरैक्शन को बाहर करने के लिए, कोई डेटा के रूप में एक ही मशीन पर सेवा उदाहरणों को तैनात करने पर विचार कर सकता है। लेकिन, एक नियम के रूप में, एक Hadoop क्लस्टर में मशीनों की संख्या काफी बड़ी है, इसलिए यह उनमें से प्रत्येक पर एक मॉडल उदाहरण को तैनात करने के लिए लाभहीन होगा, खासकर बड़े मॉडल के मामले में।

इसके अलावा, एक नियम के रूप में, एक Hadoop क्लस्टर पर, कई कार्य एक साथ किए जाते हैं जो इसके संसाधनों के लिए प्रतिस्पर्धा करते हैं, इसलिए एक अतिरिक्त सेवा के लॉन्च से क्लस्टर के समग्र प्रदर्शन में कमी आएगी।

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