स्टॉर्म फ्रेमवर्क सीखना। भाग I

2011 में, ट्विटर ने लॉन्च किया, एक्लिप्स पब्लिक लाइसेंस के तहत, स्टॉर्म डिस्ट्रीब्यूटेड कंप्यूटिंग प्रोजेक्ट। तूफान BackType पर बनाया गया था और खरीद के बाद ट्विटर पर स्विच कर दिया गया था।

स्टॉर्म एक प्रणाली है जो अपाचे हडोप के समान बड़े डेटा धाराओं के वितरित प्रसंस्करण पर केंद्रित है, लेकिन वास्तविक समय में।

तूफान की मुख्य विशेषताएं:

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

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

तूफान के तत्व

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

धारा
टपल की अनुक्रम। टपल में खेतों के लिए एक नामकरण योजना शामिल है।

टोंटी
स्ट्रीम के लिए डेटा प्रदाता। यह बाहरी स्रोतों से डेटा प्राप्त करता है, उनसे एक टपल बनाता है और इसे स्ट्रीम में भेजता है। Tuple को कई अलग-अलग स्ट्रीम में भेज सकते हैं। लोकप्रिय मैसेजिंग सिस्टम के लिए तैयार हैं: RabbitMQ / AMQP , Kestrel , JMS , Kafka ।

पेंच
डाटा हैंडलर। इनपुट ट्यूपल है। आउटपुट पर 0 या अधिक टपल भेजता है।

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

उदाहरण का उपयोग करें

कार्य

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

कार्यान्वयन

आरंभ करने के लिए, बस BasicApp इनपुट का प्रिंट आउट लें ।

एक नई टोपोलॉजी बनाएं:

TopologyBuilder builder = new TopologyBuilder(); 

एक स्पाउट CdrSpout जनरेटिंग इनपुट जोड़ें:

 builder.setSpout("CdrReader", new CdrSpout()); 

बोल्ट को दो धाराओं के साथ जोड़ें और इंगित करें कि आउटपुट स्ट्रीम CdrReader है। shuffleGrouping का अर्थ है कि CdrReader का डेटा बेतरतीब ढंग से चयनित PrintOutBolt को खिलाया जाता है।

 builder.setBolt("PrintOutBolt", new PrintOutBolt(), 2).shuffleGrouping("CdrReader"); 

कॉन्फ़िगर करें और स्थानीय स्टॉर्म क्लस्टर शुरू करें:

 Config config = new Config(); //     config.setDebug(false); LocalCluster cluster = new LocalCluster(); //   Storm  cluster.submitTopology("T1", config, builder.createTopology()); //  Topology Thread.sleep(1000*10); cluster.shutdown(); //   

आउटपुट लगभग निम्नलिखित है:

वर्ग कोष्ठक में संख्या थ्रेड आईडी है, यह देखा जा सकता है कि प्रसंस्करण समानांतर में आयोजित किया जाता है।

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

ऐसा करने के लिए, तूफान CustomStreamGrouping इंटरफ़ेस प्रदान करता है।
CdrGrouper को प्रोजेक्ट में जोड़ें। इसका कार्य टुपल को समान स्रोत संख्याओं के साथ एक ही बोल्ट पर भेजना है। ऐसा करने के लिए, CustomStreamGrouping दो कॉल प्रदान करता है:
तैयार - पहले उपयोग से पहले बुलाया:

 @Override public void prepare(WorkerTopologyContext workerTopologyContext, GlobalStreamId globalStreamId, List<Integer> integers) { tasks = new ArrayList<>(integers); //   Bolts } 

और चुनिंदा कार्य - जहां ट्यूपल से एक सूची इनपुट को खिलाया जाता है, और एक सूची जिसमें टुपल सूची में प्रत्येक स्थिति के लिए बोल्ट संख्याओं को शामिल किया गया है:

 @Override public List<Integer> chooseTasks(int i, List<Object> objects) { List<Integer> rvalue = new ArrayList<>(objects.size()); for(Object o: objects) { Cdr cdr = (Cdr) o; rvalue.add(tasks.get(Math.abs(cdr.getCallSource().hashCode()) % tasks.size())); } return rvalue; } 

बदलें श्रगग्रुपिंग को CdrGrouper BasicGroupApp के साथ बदलें:

 builder.setBolt("PrintOutBolt", new PrintOutBolt(), 2). customGrouping("CdrReader", new CdrGrouper()); 

भागो और सुनिश्चित करें कि यह इरादा के अनुसार काम करता है:

अगला, परियोजना में जोड़ें:
ClientIdBolt - सोर्स नंबर द्वारा क्लाइंट आईडी की पहचान करता है।
ClientIdGrouper - क्लाइंट आईडी द्वारा समूह ।
RaterBolt - बिलिंग में लगे हुए हैं।
CalcApp कार्यक्रम का अंतिम संस्करण है।

यदि विषय दिलचस्प है, तो अगले भाग में मुझे डेटा हानि के खिलाफ सुरक्षा तंत्र के बारे में बात करने की उम्मीद है और एक वास्तविक क्लस्टर पर चल रहा है। कोड GitHub पर उपलब्ध है ।

पुनश्च। बेशक, आपने गीत से शब्दों को नहीं निकाला है, लेकिन डेटा प्रोसेसर "बोल्ट" का नाम कुछ भ्रामक है :)

युपीडी। लेख का दूसरा भाग प्रकाशित हुआ है।