धाराओं के साथ रेडिस में समय श्रृंखला

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

धाराएं:नई Redis डेटा संरचना

Redis Streams कुंजी-मान युग्मों की केवल-परिशिष्ट समय-श्रृंखला का प्रतिनिधित्व करती हैं।

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

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

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

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

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

किसी स्ट्रीम में डेटा भेजना

स्ट्रीम केवल संलग्न हैं, इसलिए आमतौर पर आपको केवल एक ही विवरण की आवश्यकता होती है, वह है रेडिस कुंजी का नाम जिसे आप लिखना चाहते हैं, और आपके कुंजी-मूल्य जोड़े का सेट।

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

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

> 1527974818120-0

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

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

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

साधारण पठन

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

redis.xrange('site:pdx',
             '1527974818120-0',
             '+',
             'COUNT', 5).then(function(resp) {

  // resp now holds 5 readings, pass them to the open graph:
  // 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' ] ] ]

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

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

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

redis.xread('BLOCK', 10000,
            'STREAMS', 'site:pdx', '$').then(function(resp) {

  // close the windows if aqi > 50
  console.log(resp);
});

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

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

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

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

आप एक साथ कई अलग-अलग स्ट्रीम में पढ़ सकते हैं, डेटा स्वीकार करने वाली पहली स्ट्रीम से मान लौटाते हुए:

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

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

उपभोक्ता समन्वय

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

स्ट्रीम कमांड के लिए क्विक-रेफरेंस

स्ट्रीम के लिए बहुत सारे नए रेडिस कमांड उपलब्ध हैं! यहां एक त्वरित संदर्भ दिया गया है, आप आधिकारिक दस्तावेज़ीकरण में अधिक विवरण पा सकते हैं।

सरल आदेश:

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

उपभोक्ता समूह आदेश:

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

यूज केस

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

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

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

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