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

पहले भाग ने स्टॉर्म की मूल अवधारणाओं की जांच की।

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

अगला, हम स्टॉर्म में लागू डेटा हानि सुरक्षा तंत्र को देखेंगे।

मूल उदाहरण

टोंटी

अगर यह हमारे लिए महत्वपूर्ण नहीं है कि क्या टपल को संसाधित करते समय त्रुटियां थीं, तो टोंटी ने टेटल को स्पाउट ऑउटपुटकॉलिक्टर को एमिट (नई वैल्यू (...)) विधि से कॉल करके भेजा।

यदि हम यह पता लगाना चाहते हैं कि क्या टुपल को सफलतापूर्वक संसाधित किया गया था, तो कॉल एमिट (नया मान (...), msgId) की तरह दिखाई देगा, जहां msgId एक मनमाना वर्ग का एक ऑब्जेक्ट है। इस स्थिति में, ISpout इंटरफ़ेस तरीके प्रदान करता है:

• ack (Object msgId) - अगर ट्यूपल को संसाधित किया जाता है तो उसे कॉल किया जाएगा
• फेल (Object msgId) - अगर ट्यूपल को संसाधित नहीं किया जाता है तो उसे कॉल किया जाएगा

जहाँ msgId msgId है जिसके साथ SpoutOutputCollector.emit कहा जाता था।
FailAwareSpout उदाहरण:

public class FailAwareSpout extends BaseRichSpout { private Message[] messages; // Skipped ... private static class Message implements Serializable { private String message; private int failCount; private Message(String message) { this.message = message; } } // Skipped ... @Override public void nextTuple() { // Skipped ... //  Tuple c msgId outputCollector.emit(new Values(messages[messageId].message), messageId); } // Tuple   @Override public void ack(Object msgId) { Message m = messages[(Integer) msgId]; System.out.println("IN>> [" + Thread.currentThread().getId() + "] message " + m.message + " processed successfully"); } // Tuple   @Override public void fail(Object msgId) { Message m = messages[(Integer) msgId]; if(++m.failCount > MAX_RETRY_COUNT) { throw new IllegalStateException("Too many message processing errors"); } System.out.println("IN>> [" + Thread.currentThread().getId() + "] message " + m.message + " processing failed " + "[" + m.failCount + "]"); //       sendQueue.addLast((Integer) msgId); } } 

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

पेंच

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

• ack (टुपल) - प्रसंस्करण सफल रहा
• असफल (टपल) - प्रसंस्करण विफल रहा

FailingBolt उदाहरण:

 public class FailingBolt extends BaseRichBolt { OutputCollector outputCollector; // Skipped ... @Override public void execute(Tuple tuple) { // Skipped ... outputCollector.ack(tuple); //    } else { // Skipped ... outputCollector.fail(tuple); //     } } // Skipped ... } 

उपयोग का उदाहरण: BasicFailApp , Spout FailAwareSpout और बोल्ट FailingBolt बेतरतीब ढंग से प्रसंस्करण त्रुटियों को पैदा करते हैं।

BaseBasicBolt वर्ग से विरासत में मिली बोल्ट में, एक्ज (टपल) को निष्पादन विधि से बाहर निकलने के बाद स्वचालित रूप से कहा जाता है।

एंकरिंग

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

 public class MultiplierBolt extends BaseRichBolt { // Skipped ... @Override public void execute(Tuple tuple) { //    Tuple    for(int i = 0; i < MULTI_COUNT; ++i) { // Anchoring,   Tuple   outputCollector.emit(tuple, new Values(tuple.getString(0) + " - " + i)); } outputCollector.ack(tuple); } // Skipped ... } 

एंकरिंग उदाहरण: TreeFailApp

BaseBasicBolt वर्ग से विरासत में मिली बोल्ट में , निष्पादन विधि (Tuple, BasicOutputCollector) को BasicOutputCollector कलेक्टर के साथ कहा जाता है। BasicOutputCollector की विशेषता यह है कि यह स्वचालित रूप से उत्सर्जन करने पर इनपुट ट्यूपल पर एंकर बनाता है।

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

कोड GitHub पर उपलब्ध है ।

अगला भाग Transactional डेटा स्रोतों के साथ संयोजन में उपयोग किए जाने वाले Transactional टोपोलॉजी पर केंद्रित होगा।

युपीडी। लेख का अंतिम भाग प्रकाशित हुआ है।