MongoDB चेंज स्ट्रीम के साथ रीयल टाइम डेटा स्ट्रीमिंग

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

परिवर्तन धाराओं को परिभाषित करना

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

आप $changeStream ऑपरेटर और वॉच () पद्धति का उपयोग करके किसी भी अन्य सामान्य एकत्रीकरण ऑपरेटरों की तरह किसी भी संग्रह पर परिवर्तन धाराओं को परिभाषित कर सकते हैं। आप MongoCollection.watch() पद्धति का उपयोग करके परिवर्तन स्ट्रीम को भी परिभाषित कर सकते हैं।

उदाहरण

db.myCollection.watch()

स्ट्रीम सुविधाएं बदलें

• फ़िल्टरिंग परिवर्तन

  आप केवल कुछ लक्षित डेटा के लिए ईवेंट सूचनाएं प्राप्त करने के लिए परिवर्तनों को फ़िल्टर कर सकते हैं।

  उदाहरण:

  pipeline = [
     {
       $match: { name: "Bob" }
     } ];
  changeStream = collection.watch(pipeline);

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

• स्ट्रीम बदलें फिर से शुरू करें

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

• आदेशित परिवर्तन धाराएँ

  MongoDB किसी भी क्लस्टर के सभी प्रतिकृतियों और शार्क में सभी परिवर्तन स्ट्रीम घटनाओं को ऑर्डर करने के लिए एक वैश्विक तार्किक घड़ी का उपयोग करता है, इसलिए रिसीवर को हमेशा उसी क्रम में सूचनाएं प्राप्त होंगी, जिस क्रम में डेटाबेस पर लागू किया गया था।

• पूरे दस्तावेज़ों वाले इवेंट

  MongoDB डिफ़ॉल्ट रूप से मेल खाने वाले दस्तावेज़ों का हिस्सा लौटाता है। लेकिन, आप एक पूर्ण दस्तावेज़ प्राप्त करने के लिए परिवर्तन स्ट्रीम कॉन्फ़िगरेशन को संशोधित कर सकते हैं। ऐसा करने के लिए, विधि देखने के लिए { fullDocument:"updateLookup"} पास करें।
  उदाहरण:

  collection = db.collection("myColl")
  changeStream = collection.watch({ fullDocument: “updateLookup”})

• टिकाऊपन

  परिवर्तन धाराएँ केवल उस डेटा के लिए सूचित करेंगी जो अधिकांश प्रतिकृतियों के लिए प्रतिबद्ध हैं। यह सुनिश्चित करेगा कि संदेश स्थायित्व सुनिश्चित करने वाले बहुसंख्यक दृढ़ता डेटा द्वारा ईवेंट उत्पन्न होते हैं।

• सुरक्षा/पहुंच नियंत्रण

  परिवर्तन धाराएँ बहुत सुरक्षित हैं। उपयोगकर्ता केवल उन संग्रहों पर परिवर्तन धाराएँ बना सकते हैं जिन पर उनके पास पढ़ने की अनुमति है। आप उपयोगकर्ता भूमिकाओं के आधार पर परिवर्तन धाराएँ बना सकते हैं।

परिवर्तन धाराओं का उदाहरण

इस उदाहरण में, जब कोई स्टॉक मूल्य किसी सीमा से ऊपर जाता है, तो हम अधिसूचित होने के लिए स्टॉक संग्रह पर परिवर्तन धाराएँ बनाएंगे।

• क्लस्टर सेट करें

  परिवर्तन धाराओं का उपयोग करने के लिए, हमें पहले प्रतिकृति सेट बनाना होगा। सिंगल नोड रेप्लिका सेट बनाने के लिए निम्न कमांड चलाएँ।

  mongod --dbpath ./data --replSet “rs”

• स्टॉक संग्रह में कुछ रिकॉर्ड डालें

  var docs = [
   { ticker: "AAPL", price: 210 },
   { ticker: "AAPL", price: 260 },
   { ticker: "AAPL", price: 245 },
   { ticker: "AAPL", price: 255 },
   { ticker: "AAPL", price: 270 }
  ];
  db.Stocks.insert(docs)

• नोड परिवेश सेटअप करें और निर्भरताएं स्थापित करें

  mkdir mongo-proj && cd mongo-proj
  npm init -y
  npm install mongodb --save

• परिवर्तनों के लिए सदस्यता लें

  एक index.js फ़ाइल बनाएँ और उसमें निम्न कोड डालें।

  const mongo = require("mongodb").MongoClient;
  mongo.connect("mongodb://localhost:27017/?replicaSet=rs0").then(client => {
   console.log("Connected to MongoDB server");
   // Select DB and Collection
   const db = client.db("mydb");
   const collection = db.collection("Stocks");
   pipeline = [
     {
       $match: { "fullDocument.price": { $gte: 250 } }
     }
   ];
   // Define change stream
   const changeStream = collection.watch(pipeline);
   // start listen to changes
   changeStream.on("change", function(event) {
     console.log(JSON.stringify(event));
   });
  });

  अब इस फाइल को रन करें:

  node index.js

• अपडेट प्राप्त करने के लिए db में एक नया रिकॉर्ड डालें

  db.Stocks.insert({ ticker: “AAPL”, price: 280 })

  अब अपना कंसोल जांचें, आपको MongoDB से एक अपडेट प्राप्त होगा।
  उदाहरण प्रतिक्रिया:

  {
  "_id":{
  "_data":"825C5D51F70000000129295A1004E83608EE8F1B4FBABDCEE73D5BF31FC946645F696400645C5D51F73ACA83479B48DE6E0004"},
  "operationType":"insert",
  "clusterTime":"6655565945622233089",
  "fullDocument":{
  "_id":"5c5d51f73aca83479b48de6e",
  "ticker":"AAPL",
  "Price":300
  },
  "ns":{"db":"mydb","coll":"Stocks"},
  "documentKey":{"_id":"5c5d51f73aca83479b48de6e"}
  }

यहां आप किसी संग्रह में विभिन्न प्रकार के परिवर्तनों को सुनने के लिए निम्न संचालन के साथ ऑपरेशन टाइप पैरामीटर का मान बदल सकते हैं:

• सम्मिलित करें
• बदलें (अद्वितीय आईडी को छोड़कर)
• अपडेट करें
• हटाएं
• अमान्य (जब भी Mongo अमान्य कर्सर लौटाता है)

परिवर्तन स्ट्रीम के अन्य तरीके

आप डेटाबेस और परिनियोजन के विरुद्ध परिवर्तन धाराएँ उसी तरह शुरू कर सकते हैं जैसे संग्रह के विरुद्ध। यह सुविधा MongoDB संस्करण 4.0 से जारी की गई है। डेटाबेस और परिनियोजन के विरुद्ध परिवर्तन स्ट्रीम खोलने के लिए यहां आदेश दिए गए हैं।

Against DB: db.watch()
Against deployment: Mongo.watch()

निष्कर्ष

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