🧔🏻 🧑🏿 🌩️ स्टॉर्म फ्रेमवर्क सीखना। भाग III 🗃️ 🃏 👝

लेख के दूसरे भाग में प्रसंस्करण के दौरान त्रुटियों का पता लगाने के लिए तंत्र के बारे में बात की गई थी।

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

लेनदेन की विशेषताओं के बारे में बात करते समय, ACID शब्द तुरंत पॉप अप होता है:

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

संगति और स्थायित्व डेटाबेस से अधिक संबंधित हैं। हम एटमॉसिटी और अलगाव में रुचि लेंगे।

0.8.0 संस्करण में, स्टॉर्म ने ट्राइडेंट सबसिस्टम - अपाचे सुअर का एक एनालॉग पेश किया। ट्रांसेक्शनल टोपोलॉजी फंक्शनल ने इसे माइग्रेट किया।

तूफान में लेनदेन

एक टोमिसिटी

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

मैं विलाप करता हूं

स्टॉर्म गारंटी देता है कि ट्रांजैक्शन आईडी के आरोही क्रम में बैच को स्टेटयूपेडर में सख्ती से क्रमिक रूप से प्रेषित किया जाता है। यानी बैच # 1 के सफल निर्धारण के बाद ही बैच # 2 तय किया जाएगा।

कार्यान्वयन

एक ट्रांजेक्शनल स्पाउट को ICommitterTridentSpout <TransactionMetadata> इंटरफ़ेस को लागू करना चाहिए। TransactionMetadata - कोई भी वर्ग जिसमें बैच उत्पन्न करने और निम्नलिखित लेनदेन उत्पन्न करने के लिए डेटा होता है: TxMeta ।

छिपा हुआ पाठ

public class TxMeta { private int start; private int count; public TxMeta(int start, int count) { this.start = start; this.count = count; } // Skipped getters }

वह वर्ग जो ITridentSpout.BatchCoordinator <TransactionMetadata> इंटरफ़ेस को लागू करता है, लेन-देन बनाते समय TransactionMetadata को आरंभ करता है और अनुरोध का जवाब देता है कि क्या डेटा अगले लेनदेन के लिए तैयार है - TridentTxSpout । यह प्रत्येक टोपोलॉजी के लिए एक ही प्रति में बनाया गया है।

छिपा हुआ पाठ

  static class BCoordinator implements BatchCoordinator<TxMeta> { private static final int TRANSACTION_COUNT = 5; private static final int TRANSACTION_ELEMENT_COUNT = 5; //TxMeta -    @Override public TxMeta initializeTransaction(long l, TxMeta txMeta) { if(txMeta != null) { System.out.println(String.format("Initializing transaction id: %08d, " + "start: %04d, count: %04d", l, txMeta.getStart() + txMeta.getCount(), txMeta.getCount())); return new TxMeta(txMeta.getStart() + txMeta.getCount(), TRANSACTION_ELEMENT_COUNT); } else { return new TxMeta(0, TRANSACTION_ELEMENT_COUNT); } } //       @Override public boolean isReady(long l) { if(l <= TRANSACTION_COUNT) { System.out.println("ISREADY " + l); return true; } return false; } }

वह वर्ग जो ICommitterTridentSpout.Emitter इंटरफ़ेस लागू करता है एक बैच बनाता है। बैच को संसाधित करने में त्रुटि के मामले में, फिर से बैच बनाता है।
महत्वपूर्ण - एक पुनर्निर्मित बिल्ट में मूल रूप में एक ही टपल सेट होना चाहिए।

छिपा हुआ पाठ

 static class BEmitter implements Emitter { //  Batch    TransactionMetadata @Override public void emitBatch(TransactionAttempt transactionAttempt, Object coordinatorMeta, TridentCollector tridentCollector) { TxMeta txMeta = (TxMeta) coordinatorMeta; System.out.println("Emitting transaction id: " + transactionAttempt.getTransactionId() + " attempt:" + transactionAttempt.getAttemptId() ); for(int i = 0; i < txMeta.getCount(); ++i) { tridentCollector.emit(new Values("TRANS [" + transactionAttempt.getAttemptId() + "] [" + (txMeta.getStart() + i) + "]") ); } } //     State @Override public void success(TransactionAttempt transactionAttempt) { System.out.println("BEmitter:Transaction success id:" + transactionAttempt.getTransactionId()); } //     State @Override public void commit(TransactionAttempt transactionAttempt) { System.out.println("BEmitter:Transaction commit id:" + transactionAttempt.getTransactionId()); } }

हमारे मामले में राज्य इंटरफ़ेस को लागू करने वाला वर्ग, डेटाबेस ड्राइवर: TxDatabase ।

छिपा हुआ पाठ

 public class TxDatabase implements State { //       @Override public void beginCommit(Long txId) { System.out.println("beginCommit [" + Thread.currentThread().getId() + "] " + txId); } //       @Override public void commit(Long txId) { System.out.println("commit [" + Thread.currentThread().getId() + "] " + txId); } }

BaseStateUpdater को विरासत में मिला वर्ग <S राज्य का विस्तार करता है, राज्य में परिवर्तन करता है (DB): TxDatabaseUpdater

छिपा हुआ पाठ

 public class TxDatabaseUpdater extends BaseStateUpdater<TxDatabase> { int count; //     @Override public void updateState(TxDatabase txDatabase, List<TridentTuple> tridentTuples, TridentCollector tridentCollector) { //    if(++count == 2) throw new FailedException("YYYY"); for(TridentTuple t: tridentTuples) { System.out.println("Updating: " + t.getString(0)); } } }

एक वर्ग जो स्टेटफैक्टरी इंटरफेस को लागू करता है, वह स्टेट को त्वरित करता है: TxDatabaseFactory ।

यह सब एक साथ रखकर TridentTransactionApp :

 public class TridentTransactionApp { public static void main( String[] args ) throws Throwable { Logger.getRootLogger().setLevel(Level.ERROR); //   TridentTopology tridentTopology = new TridentTopology(); //   Spout tridentTopology.newStream("TridentTxSpout", new TridentTxSpout()). //  Tuple   - OpPrintout    shuffle().each(new Fields("msg"), new OpPrintout()). parallelismHint(2). //        global(). //    State () partitionPersist(new TxDatabaseFactory(), new Fields("msg"), new TxDatabaseUpdater()); // Skipped LocalCluster cluster = new LocalCluster(); cluster.submitTopology("T2", config, tridentTopology.build()); Thread.sleep(1000*100); cluster.shutdown(); } }

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

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

पी पी एस। Hadoop में, Storm के ऑन-लाइन प्रोसेसिंग का एक एनालॉग है - Hadoop Streaming , लेकिन Storm के विपरीत, यह लेनदेन का समर्थन नहीं करता है।

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

तूफान में लेनदेन

एक टोमिसिटी

मैं विलाप करता हूं

कार्यान्वयन

More articles: