👎🏻 🤼 🌑 मनमानी मशीन लर्निंग मॉडल का उपयोग करके अपाचे स्पार्क पर बड़ी मात्रा में डेटा को वर्गीकृत करें 🌺 👩🏻‍🔧 👩🏿‍🤝‍👩🏼

भाग 1: समस्या कथन

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

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

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

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

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

बेशक, कई मामलों में, नेटवर्क इंटरैक्शन को पूरी तरह से टाला नहीं जा सकता है, लेकिन, फिर भी, आपको स्थानीय रूप से सभी कार्यों को करने और नेटवर्क पर संचारित होने की आवश्यकता वाले डेटा की मात्रा को कम करने की कोशिश करने की आवश्यकता है।

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

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

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

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

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

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

val model: LogisticRegressionModel = LogisticRegressionModel.load("/path/to/model") val dataset = spark.read.parquet("/path/to/data") val result = model.transform(dataset)

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

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

चित्र 2. एक सेवा के रूप में मॉडल की अवधारणा

पायथन वर्गीकरण सेवा विवरण

मॉडल को सेवा के रूप में प्रस्तुत करने के लिए, निम्नलिखित कार्यों को हल करना आवश्यक है:

HTTP के माध्यम से मॉडल तक कुशल पहुंच को लागू करना;
मशीन संसाधनों (मुख्य रूप से सभी प्रोसेसर कोर और मेमोरी) का सबसे कुशल उपयोग सुनिश्चित करें;
उच्च भार के लिए प्रतिरोध प्रदान करते हैं;
क्षैतिज रूप से स्केल करने की क्षमता प्रदान करते हैं।

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

 from flask import Flask, request, Response model = load_model() n_features = 100 app = Flask(__name__) @app.route("/score", methods=['PUT']) def score(): inp = np.frombuffer(request.data, dtype='float32').reshape(-1, n_features) result = model.predict(inp) return Response(result.tobytes(), mimetype='application/octet-stream') if __name__ == "__main__": app.run()

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

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

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

 PUT /score HTTP/1.1 Host: score-node-1:8099 User-Agent: curl/7.58.0 Accept: */* Content-Type: application/binary Content-Length: 400 [400 bytes of data]

जैसा कि आप देख सकते हैं, इस तरह के अनुरोध की दक्षता बहुत कम है (पेलोड के 400 बाइट्स / (133 बाइट्स हेडर + 400 बाइट्स बॉडी) = 75%)। सौभाग्य से, लगभग सभी पुस्तकालयों में, पूर्वानुमान पद्धति आपको [1 xn] वेक्टर नहीं, बल्कि [mxn] मैट्रिक्स प्राप्त करने की अनुमति देती है, और, तदनुसार, मी इनपुट मानों के लिए तुरंत परिणाम प्राप्त करती है।

इसके अलावा, बड़े मैट्रिस के साथ काम करने के लिए सुपी लाइब्रेरी को अनुकूलित किया गया है, जिससे आप सभी उपलब्ध मशीन संसाधनों का प्रभावी ढंग से उपयोग कर सकते हैं। इस प्रकार, हम एक अनुरोध में एक नहीं बल्कि बड़ी संख्या में फ़ीचर वैक्टर भेज सकते हैं, उन्हें आकार के एक सूक्ष्म मैट्रिक्स [mxn] में वर्गीकृत कर सकते हैं, वर्गीकृत कर सकते हैं और वेक्टर [mx 1] को बूलियन या flox32 मानों से वापस कर सकते हैं। परिणामस्वरूप, 1000 पंक्तियों के मैट्रिक्स का उपयोग करते समय HTTP इंटरैक्शन की दक्षता लगभग 100% के बराबर हो जाती है। इस मामले में HTTP हेडर का आकार उपेक्षित किया जा सकता है।

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

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

 server { listen 80; server_name 127.0.0.1; location / { try_files $uri @score; } location @score { include uwsgi_params; uwsgi_pass unix:/tmp/score.sock; } }

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

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

 FROM ubuntu #Installing required ubuntu and python modules RUN apt-get update RUN apt-get -y install python3 python3-pip nginx RUN update-alternatives --install /usr/bin/python python /usr/bin/python3 1 RUN update-alternatives --install /usr/bin/pip pip /usr/bin/pip3 1 RUN pip install uwsgi flask scipy scikit-learn #copying script files WORKDIR /etc/score COPY score.py . COPY score.ini . COPY start.sh . RUN chmod +x start.sh RUN rm /etc/nginx/sites-enabled/default COPY score.nginx /etc/nginx/sites-enabled/ EXPOSE 80 ENTRYPOINT ["./start.sh"]

तो, वर्गीकरण के लिए सेवा की संरचना इस प्रकार है:

चित्रा 3. वर्गीकरण के लिए सेवा योजना

हाडोप इकोसिस्टम में अपाचे स्पार्क के काम का एक संक्षिप्त सारांश

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

ये सभी कार्य Hadoop पारिस्थितिक तंत्र के साथ काम करने वाले विभिन्न ढाँचों द्वारा पूरे किए जाते हैं। सबसे लोकप्रिय और सुविधाजनक अपाचे स्पार्क में से एक है। मुख्य अवधारणा जिसके चारों ओर पूरा ढांचा बनाया गया है, वह है आरडीडी (रेजिलिएंट डिस्ट्रीब्यूटेड डेटसेट)। सामान्य तौर पर, RDD को एक वितरित संग्रह के रूप में माना जा सकता है जो ड्रॉप-प्रतिरोधी है। RDD को दो मुख्य तरीकों से प्राप्त किया जा सकता है:

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

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

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

RDD स्पार्क 1.x के संस्करणों में स्पार्क एपीआई के साथ बातचीत की मुख्य इकाई थी। स्पार्क 2.x में, डेवलपर्स ने कहा कि अब बातचीत के लिए मुख्य अवधारणा डेटासेट है। डेटासेट SQL-जैसे इंटरैक्शन के समर्थन के साथ RDD के लिए एक ऐड-ऑन है। डेटासैट एपीआई का उपयोग करते समय, स्पार्क आपको काफी निम्न-स्तरीय लोगों सहित अनुकूलन की एक विस्तृत श्रृंखला का उपयोग करने की अनुमति देता है। लेकिन सामान्य तौर पर, मूल सिद्धांत जो RDD पर लागू होते हैं, वे भी डेटासेट पर लागू होते हैं।

स्पार्क के काम के बारे में अधिक विवरण आधिकारिक वेबसाइट पर प्रलेखन में पाया जा सकता है।

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

 case class Data(id: String, text: String) case class Features(id: String, vector: Array[Float]) case class Score(id: String, score: Float) //(1) def std(vector: Array[Float]): Float = ??? //(2) val ds: Dataset[Data] = spark.read.parquet("/path/to/data").as[Data] //(3) val result: Dataset[Score] = ds.map {d: Data => //(4) val filteredText = d.text.toLowerCase.filter { letter => 'a' <= letter && letter <= 'z' } val featureVector = new Array[Float](26) if (filteredText.nonEmpty) { filteredText.foreach(letter => featureVector(letter) += 1) featureVector.indicies.foreach { i => featureVector(i) = featureVector(i) / filteredText.length() } } Features(d.id, featureVector) }.map {f: Features => Score(f.id, std(f.vector)) //(5) } result.write.parquet("/path/to/result") //(6)

इस उदाहरण में, हम:

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

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

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

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

 case class Data(id: String, text: String) case class Features(id: String, vector: Array[Float]) case class Score(id: String, score: Float) def std(vector: Array[Float]): Float = ??? val ds: Dataset[Data] = spark.read.parquet("/path/to/data").as[Data] val result: Dataset[Score] = ds.map { // --------------- EXECUTOR PART BEGIN ----------------------- d: Data => val filteredText = d.text.toLowerCase.filter { letter => 'a' <= letter && letter <= 'z' } val featureVector = new Array[Float](26) if (filteredText.nonEmpty) { filteredText.foreach(letter => featureVector(letter) += 1) featureVector.indicies.foreach { i => featureVector(i) = featureVector(i) / filteredText.length() } } Features(d.id, featureVector) // --------------- EXECUTOR PART END ----------------------- }.map { // --------------- EXECUTOR PART BEGIN ----------------------- f: Features => Score(f.id, std(f.vector)) // --------------- EXECUTOR PART END ----------------------- } result.write.parquet(“/path/to/result”)

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

स्पार्क एप्लिकेशन को लॉन्च करते समय, YARN एक क्लस्टर मशीन पर कंटेनर बनाता है और चलाता है जिसमें यह चालक को लॉन्च करता है। फिर, जब चालक संचालन से डीएजी तैयार करता है जिसे निष्पादकों पर चलाने की आवश्यकता होती है, तो YARN वांछित मशीनों पर अतिरिक्त कंटेनर लॉन्च करता है।

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

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

अब देखते हैं कि हमारे उदाहरण से कोड को सीधे क्लस्टर पर कैसे निष्पादित किया जाएगा। कल्पना कीजिए कि स्रोत डेटा का आकार 2 टेराबाइट्स है। तदनुसार, अगर एचडीएफएस पर ब्लॉक का आकार 128 मेगाबाइट है, तो कुल 16384 ब्लॉक होंगे। प्रत्येक ब्लॉक विश्वसनीयता सुनिश्चित करने के लिए कई मशीनों को दोहराया जाता है। सरलता के लिए, हम प्रतिकृति कारक को दो के बराबर लेते हैं, अर्थात्, कुल मिलाकर 32768 उपलब्ध ब्लॉक होंगे। मान लीजिए कि हम भंडारण के लिए 16 मशीनों के एक क्लस्टर का उपयोग करते हैं। तदनुसार, समान वितरण के मामले में प्रत्येक मशीन पर लगभग 2048 ब्लॉक, या प्रति मशीन 256 गीगाबाइट होंगे। प्रत्येक मशीन पर, हमारे पास 8 प्रोसेसर कोर और 64 गीगाबाइट रैम है।

हमारे कार्य के लिए, ड्राइवर को बहुत अधिक संसाधनों की आवश्यकता नहीं है, इसलिए हम इसके लिए 1 कोर और 1 जीबी मेमोरी आवंटित करेंगे। हम कलाकारों को 2 कोर और 4 जीबी मेमोरी देंगे। मान लीजिए हम क्लस्टर संसाधनों का उपयोग अधिकतम करना चाहते हैं। इस प्रकार, हम 64 कंटेनरों को प्राप्त करते हैं: चालक के लिए एक, और कलाकारों के लिए 63।

चित्रा 4. डेटा नोड और उनके द्वारा उपयोग किए जाने वाले संसाधनों पर चलने वाली प्रक्रियाएं।

चूंकि हमारे मामले में हम केवल मानचित्र संचालन का उपयोग करते हैं, हमारे डीएजी में एक ऑपरेशन शामिल होगा। इसमें निम्नलिखित क्रियाएं शामिल हैं:

स्थानीय हार्ड ड्राइव से डेटा का एक ब्लॉक लें,
डेटा कन्वर्ट करें
अपने स्थानीय डिस्क पर नए ब्लॉक के परिणाम को सहेजें।

कुल में, हमें 16384 ब्लॉकों को संसाधित करने की आवश्यकता है, इसलिए प्रत्येक निष्पादक को 16384 / (63 निष्पादक * 2 कोर) = 130 संचालन करना होगा। इस प्रकार, निष्पादक का जीवन चक्र एक अलग प्रक्रिया के रूप में (यदि सब कुछ बिना गिर जाए) निम्नानुसार दिखेगा।

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

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

अपाचे स्पार्क और बाहरी सेवाओं के बीच बातचीत की मुख्य समस्याएं

यदि, मानचित्र संचालन के ढांचे के भीतर, हमें कुछ बाहरी सेवा तक पहुंचने की आवश्यकता है, तो कार्य कम तुच्छ हो जाता है। मान लीजिए कि बाहरी सेवा के साथ बातचीत करने के लिए बाहरी सेवा का एक वर्ग जिम्मेदार है। सामान्य तौर पर, काम शुरू करने से पहले, हमें इसे शुरू करने की आवश्यकता है, और फिर इसे आवश्यक रूप से कॉल करें:

 val client = ExternalServiceClient.create() // val score = client.score(featureVector) // .

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

स्पार्क में ऐसा करने के दो तरीके हैं। सबसे पहले, यदि ग्राहक क्रमबद्ध है (क्लाइंट स्वयं और उसके सभी फ़ील्ड java.io.Serializable इंटरफ़ेस का विस्तार करना चाहिए), तो इसे ड्राइवर पर आरंभीकृत किया जा सकता है और फिर प्रसारण चर तंत्र के माध्यम से निष्पादकों को दिया जा सकता है।

 val client = ExternalServiceClient.create() val clientBroadcast = sparkContext.broadcast(client) ds.map { f: Features => val score = clientBroadcast.value.score(f.vector) Score(f.id, score) }

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

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

 ds.mapPartitions {fi: Iterator[Features] => val client = ExternalServiceClient.create() fi.map { f: Features => val score = client.score(f.vector) Score(f.id, score) } }

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

परिणामी पुनरावृत्ति का प्रसंस्करण संचालन एकल-थ्रेडेड है। आपको याद दिला दूं कि इट्रेटर टाइप स्ट्रक्चर को एक्सेस करने का मुख्य पैटर्न hasNext और अगली विधियों के लिए एक क्रमिक कॉल है:

 while (i.hasNext()) { val item = i.next() … }

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

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

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

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

चित्र 6. तत्वों के एक समूह के लिए एक वर्गीकरण अनुरोध भेजते समय सिंक्रोनस बातचीत

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

समस्या का अंतिम विवरण

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

हालांकि, इस तरह के अतुल्यकालिक प्रसंस्करण कई अतिरिक्त समस्याएं प्रस्तुत करते हैं।

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

इन समस्याओं को प्रभावी ढंग से हल करने में हम कैसे सक्षम थे, इसके बारे में मैं अगले भाग में बताऊंगा । देखते रहो!

इस बीच, हमारी कंपनी की रिक्तियों को देखें, शायद हम आपको ढूंढ रहे हैं?

Java-

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

भाग 1: समस्या कथन

पायथन वर्गीकरण सेवा विवरण

हाडोप इकोसिस्टम में अपाचे स्पार्क के काम का एक संक्षिप्त सारांश

अपाचे स्पार्क और बाहरी सेवाओं के बीच बातचीत की मुख्य समस्याएं

समस्या का अंतिम विवरण

More articles: