Redis Streams का उपयोग करके टाइम सीरीज़ डेटा को मॉडल करना

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

Streams: Redis का नया डेटा स्ट्रक्चर

Redis Streams की-वैल्यू पेयर्स का एक ऐपेंड-ओनली टाइम सीरीज़ प्रस्तुत करते हैं।

कोई भी संख्या में क्लाइंट्स एक Stream में लिख सकते हैं और हर बार जब वे ऐसा करते हैं तो उन्हें उस आइटम का एक यूनिक आरोही ID प्राप्त होता है जिसे टाइम सीरीज़ में डाला गया था।

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

Streams एक बड़ा विषय है, इसलिए यह पोस्ट केवल संक्षेप में देखेगा कि Redis में टाइम सीरीज़ डेटा को मॉडल करने के लिए उनका उपयोग कैसे किया जाए। नया डेटा स्ट्रक्चर इसे पिछले दृष्टिकोणों की तुलना में नाटकीय रूप से सरल बनाता है जैसे List या Sorted Set टाइप्स का उपयोग करके टाइम सीरीज़ को मॉडल करना।

यहां कोड उदाहरणों के लिए हम Node.js के लिए ioredis क्लाइंट का उपयोग करेंगे:

var Redis = require('ioredis'); var redis = new Redis(); 

Stream में डेटा भेजना

Streams ऐपेंड-ओनली होते हैं, इसलिए आमतौर पर आपको केवल उस Redis की का नाम चाहिए जिसे आप लिखना चाहते हैं, और की-वैल्यू पेयर्स का आपका सेट।

मैं अपने एयर क्वालिटी सेंसर से माप रिकॉर्ड करने जा रहा हूँ, विशेष रूप से वर्तमान एयर क्वालिटी इंडेक्स और सेल्सियस में तापमान को site:pdx की पर की-वैल्यू पेयर्स की एक स्ट्रीम भेज रहा हूँ:

redis.xadd('site:pdx', '*', 'aqi', 37, 'tempc', 5.1).then(function(id) { console.log("id:", id); }); > 1527974818120-0 

मैं जब भी एक नया माप रिकॉर्ड करना होता है तो एक XADD कमांड भेजता हूँ। प्रतिक्रिया एक यूनिक, हमेशा-आरोही ID होती है जिसका उपयोग क्वेरीज़ में किया जा सकता है। ID का पहला भाग, 1527974818120 Redis सर्वर द्वारा असाइन किया गया टाइमस्टैम्प है। ID का दूसरा भाग एक इन्क्रीमेंटिंग नंबर है ताकि जब कई क्लाइंट्स एक ही समय पर लिखें तो टकराव से बचा जा सके।

लिखने की कमांड में दूसरे आर्ग्यूमेंट के रूप में * सप्लाई करना, जैसा कि ऊपर दिखाया गया है, Redis को अपने स्वयं के टाइमस्टैम्प के साथ डेटा स्टोर करने देता है।

डेटा लिखते समय टाइमस्टैम्प प्रदान करना भी संभव है, लेकिन आमतौर पर यह पसंद नहीं किया जाता है: Redis को टाइमस्टैम्प चुनने देने से कई क्लाइंट्स एक ही Stream में एक साथ डेटा लिख सकते हैं बिना क्लाइंट्स को ID चयन और ऑर्डरिंग के बारे में समन्वय करने की आवश्यकता के - Redis सर्वर क्लाइंट्स के लिए उन विवरणों को संभालता है।

सरल पढ़ना

एक बार Stream में कुछ वैल्यूज़ होने के बाद आप ID या टाइमस्टैम्प की एक रेंज सप्लाई करके वैल्यूज़ की एक रेंज प्राप्त कर सकते हैं। इसका उपयोग मेरे एयर क्वालिटी मेट्रिक्स के सबसे हालिया रीडिंग को ग्राफ पर प्रदर्शित करने वाले ऐप द्वारा किया जा सकता है:

redis.xrange('site:pdx', '1527974818120-0', '+', 'COUNT', 5).then(function(resp) { // resp अब 5 रीडिंग रखता है, उन्हें ओपन ग्राफ पर पास करें: // console.log(resp); }); > [ [ '1543947167906-0', [ 'aqi', '31', 'tempc', '5.1' ] ], > [ '1543947168312-0', [ 'aqi', '31', 'tempc', '5.3' ] ], > [ '1543947168901-0', [ 'aqi', '31', 'tempc', '5.4' ] ], > [ '1543947170033-0', [ 'aqi', '31', 'tempc', '5.4' ] ], > [ '1543947171460-0', [ 'aqi', '31', 'tempc', '5.6' ] ] ] 

Stream में कहीं से भी नंबरों की रेंज को सैंपल किया जा सकता है, जिससे ग्राफिंग सिस्टम बिना प्रदर्शन दंड के ऐतिहासिक डेटा क्वेरी कर सकता है।

ब्लॉकिंग और पोलिंग

डेटा की रेंज क्वेरी करना ग्राफ और ऐतिहासिक मॉनिटरिंग के लिए उपयोगी है, लेकिन कभी-कभी आप एक ऐसा सिस्टम बनाना चाहते हैं जो डेटा आते ही तुरंत प्रतिक्रिया दे सके - Redis Streams इसके लिए भी बहुत उपयुक्त हैं, XREAD का उपयोग करके:

redis.xread('BLOCK', 10000, 'STREAMS', 'site:pdx', '$').then(function(resp) { // यदि aqi > 50 तो खिड़कियां बंद करें console.log(resp); }); 

इस तरह का एक ब्लॉकिंग ऑपरेशन तब तक इंतजार करेगा जब तक डेटा नहीं आता, या टाइमआउट तक (यहां 1000ms), इसलिए लगातार पोल बनाए रखने के लिए आप ऊपर दिए गए उदाहरण की तरह ब्लॉक करते हैं जब तक डेटा उपलब्ध नहीं हो जाता, और डेटा प्राप्त होने या कमांड टाइमआउट होने के बाद बस उसी XREAD कमांड को फिर से कॉल करते हैं।

अपने Stream से कनेक्शन के बीच कोई डेटा मिस न हो यह सुनिश्चित करने के लिए, आप अपने Stream से पढ़े गए अंतिम ID को सप्लाई कर सकते हैं जब आप उससे पढ़ रहे हों, और इस तरह आप ठीक वहीं से शुरू करेंगे जहां आपने छोड़ा था।

उपरोक्त मामले में मैं सिर्फ Stream पर आने वाला जो भी डेटा चाहता था इसलिए 'नए डेटा केवल' को इंगित करने के लिए विशेष टोकन '$' का उपयोग किया; सप्लाई किए गए ID के साथ कमांड केवल थोड़ा अलग दिखती है:

redis.xread('BLOCK', 10000, 'STREAMS', 'site:pdx', '1543947171460-0'); 

आप एक साथ कई अलग-अलग Streams से भी पढ़ सकते हैं, पहले Stream से वैल्यूज़ लौटाते हुए जो डेटा स्वीकार करता है:

redis.xread('BLOCK', 10000, 'STREAMS', 'site:pdx', 'site:global', '1543947171460-0', '$'); 

इस उदाहरण में कमांड तब लौटेगी जब ID 1543947171460-0 से अधिक हालिया कोई डेटा site:pdx पर लिखा जाता है, या जब कोई भी नया डेटा site:global पर लिखा जाता है।

कंज्यूमर समन्वय

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

Stream कमांड्स के लिए क्विक रेफरेंस

Streams के लिए बहुत सारे नए Redis कमांड्स उपलब्ध हैं! यहां एक क्विक रेफरेंस है, आप आधिकारिक डॉक्यूमेंटेशन में अधिक विवरण पा सकते हैं।

सरल कमांड्स:

• XADD: एक Stream में एक आइटम (की-वैल्यू पेयर्स का बंडल) जोड़ें
• XRANGE और XREVRANGE: रेंज चुनें या Stream आइटम्स पर इटरेट करें
• XREAD: कुछ ID से अधिक हालिया आइटम्स प्राप्त करें (वैकल्पिक रूप से ब्लॉकिंग)
• XTRIM: Stream को ट्रिम करने के लिए पुराने आइटम्स को हटाएं
• XDEL: Stream से एक विशिष्ट आइटम हटाएं
• XLEN: Stream में आइटम्स गिनें
• XINFO: Stream मेटाडेटा का निरीक्षण करें

कंज्यूमर ग्रुप कमांड्स:

• XGROUP: कंज्यूमर ग्रुप्स बनाएं, हटाएं, या रीसेट करें, और उनके सदस्यों को हटाएं
• XREADGROUP: XREAD की तरह (ऊपर), लेकिन कंज्यूमर ग्रुप का उपयोग करके मैसेजेस प्राप्त करें
• XPENDING: उन मैसेजेस का निरीक्षण करें जो कंज्यूमर ग्रुप को डिलीवर किए गए हैं लेकिन स्वीकार नहीं किए गए
• XACK: कंज्यूमर ग्रुप के लिए एक मैसेज स्वीकार करें, इसे पेंडिंग सूची से हटा दें
• XCLAIM: एक मृत कंज्यूमर से एक मैसेज का अधिग्रहण करें

उपयोग के मामले

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

Streams उन एप्लिकेशन्स में भी उपयोग के लिए उपयुक्त हैं जो पहले अन्य Redis डेटा स्ट्रक्चर्स का उपयोग करते थे। Sidekiq और Celery जैसे क्यूइंग एप्लिकेशन्स Streams के कंज्यूमर ग्रुप्स का लाभ उठा सकते हैं ताकि रीड रिसीप्ट्स के निरीक्षण को Redis-नेटिव तरीके से प्रदान किया जा सके। Redis pubsub का उपयोग करके सरल चैट ऐप्स दिखाने वाले बहुत सारे ब्लॉग पोस्ट हैं जिन्हें Redis Streams के साथ अधिक मजबूत बनाया जा सकता है, क्योंकि pubsub मैसेजेस को क्लाइंट्स को पब्लिश करने के बाद बरकरार नहीं रखता है।

आज़माने के लिए तैयार?

आप बस कुछ क्लिक्स के साथ एक नए RedisGreen सर्वर पर Redis Streams आज़मा सकते हैं।