RedisTimeSeries और Apache Kafka के साथ टाइम-सीरीज पाइपलाइन्स को ऑप्टिमाइज़ करना

RedisTimeSeries एक रेडिस मॉड्यूल है जो रेडिस में नेटिव टाइम-सीरीज डेटा संरचना लाता है। टाइम-सीरीज समाधान, जो पहले सॉर्टेड सेट्स (या रेडिस स्ट्रीम्स) के ऊपर बने थे, RedisTimeSeries की विशेषताओं जैसे हाई-वॉल्यूम इन्सर्ट्स, लो-लेटेंसी रीड्स, फ्लेक्सिबल क्वेरी लैंग्वेज, डाउन-सैंपलिंग, और बहुत कुछ से लाभ उठा सकते हैं!

आम तौर पर, टाइम-सीरीज डेटा (अपेक्षाकृत) सरल होता है। इसके बावजूद, हमें अन्य विशेषताओं को भी ध्यान में रखना होगा:

• डेटा वेग: जैसे सैकड़ों मेट्रिक्स हजारों डिवाइसेज से प्रति सेकंड
• वॉल्यूम (बिग डेटा): महीनों (यहां तक कि वर्षों) में डेटा संचय

इस प्रकार, RedisTimeSeries जैसे डेटाबेस समग्र समाधान का केवल एक हिस्सा हैं। आपको यह भी सोचना होगा कि कैसे अपने सभी डेटा को RedisTimeSeries में एकत्रित करें (इनजेस्ट), प्रोसेस करें, और भेजें। आपको वास्तव में एक स्केलेबल डेटा पाइपलाइन की आवश्यकता है जो उत्पादकों और उपभोक्ताओं को डीकोपल करने के लिए बफर के रूप में कार्य कर सके।

यहीं Apache Kafka आता है! कोर ब्रोकर के अलावा, इसमें घटकों का एक समृद्ध पारिस्थितिकी तंत्र है, जिसमें Kafka Connect (जो इस ब्लॉग पोस्ट में प्रस्तुत समाधान आर्किटेक्चर का एक हिस्सा है), कई भाषाओं में क्लाइंट लाइब्रेरीज, Kafka Streams, Mirror Maker, आदि शामिल हैं।

यह ब्लॉग पोस्ट टाइम-सीरीज डेटा के विश्लेषण के लिए RedisTimeSeries को Apache Kafka के साथ उपयोग करने का एक व्यावहारिक उदाहरण प्रदान करता है।

कोड इस GitHub रेपो में उपलब्ध है: https://github.com/abhirockzz/redis-timeseries-kafka

आइए पहले उपयोग के मामले का पता लगाएं। कृपया ध्यान दें कि इसे ब्लॉग पोस्ट के उद्देश्यों के लिए सरल रखा गया है और बाद के खंडों में आगे समझाया गया है।

परिदृश्य: डिवाइस मॉनिटरिंग

कल्पना करें कि कई स्थान हैं, जिनमें से प्रत्येक में कई डिवाइस हैं, और आपको डिवाइस मेट्रिक्स की निगरानी करने का काम सौंपा गया है—अभी के लिए हम तापमान और दबाव पर विचार करेंगे। ये मेट्रिक्स RedisTimeSeries में संग्रहीत किए जाएंगे (बेशक!) और कुंजियों के लिए निम्नलिखित नामकरण परंपरा का उपयोग करेंगे—<मेट्रिक नाम>:<स्थान>:<डिवाइस>। उदाहरण के लिए, स्थान 5 में डिवाइस 1 का तापमान temp:5:1 के रूप में दर्शाया जाएगा। प्रत्येक टाइम-सीरीज डेटा पॉइंट में निम्नलिखित लेबल (कुंजी-मान जोड़े) भी होंगे—मेट्रिक, स्थान, डिवाइस। यह लचीली क्वेरी की अनुमति देने के लिए है जैसा कि आप आगामी खंडों में देखेंगे।

यहां कुछ उदाहरण दिए गए हैं कि आप TS.ADD कमांड का उपयोग करके डेटा पॉइंट्स कैसे जोड़ेंगे:

# स्थान 3 में डिवाइस 2 का तापमान लेबल के साथ:

TS.ADD temp:3:2 * 20 LABELS metric temp location 3 device 2

# स्थान 3 में डिवाइस 2 का दबाव:

TS.ADD pressure:3:2 * 60 LABELS metric pressure location 3 device 2

समाधान आर्किटेक्चर

उच्च स्तर पर समाधान इस तरह दिखता है:

आइए इसे तोड़ते हैं:

स्रोत (स्थानीय) घटक

• MQTT ब्रोकर (mosquitto): MQTT IoT उपयोग के मामलों के लिए एक डी-फैक्टो प्रोटोकॉल है। हम जिस परिदृश्य का उपयोग करेंगे वह IoT और टाइम सीरीज का संयोजन है—इस पर बाद में और अधिक।
• Kafka Connect: MQTT स्रोत कनेक्टर का उपयोग MQTT ब्रोकर से Kafka क्लस्टर में डेटा स्थानांतरित करने के लिए किया जाता है।

Azure सेवाएं

• Azure Cache for Redis Enterprise टियर्स: Enterprise टियर्स Redis Enterprise पर आधारित हैं, जो Redis का एक कमर्शियल वेरिएंट है। RedisTimeSeries के अलावा, Enterprise टियर RediSearch और RedisBloom का भी समर्थन करता है। ग्राहकों को Enterprise टियर्स के लिए लाइसेंस अधिग्रहण के बारे में चिंता करने की आवश्यकता नहीं है। Azure Cache for Redis इस प्रक्रिया को सुविधाजनक बनाएगा जिसमें ग्राहक Azure Marketplace ऑफ़र के माध्यम से इस सॉफ़्टवेयर के लाइसेंस को प्राप्त और भुगतान कर सकते हैं।
• Azure पर Confluent Cloud: एक पूरी तरह से प्रबंधित पेशकश जो Apache Kafka को एक सेवा के रूप में प्रदान करती है, Azure से Confluent Cloud तक एक एकीकृत प्रोविजनिंग लेयर के लिए धन्यवाद। यह क्रॉस-प्लेटफॉर्म प्रबंधन के बोझ को कम करता है और Azure इन्फ्रास्ट्रक्चर पर Confluent Cloud का उपयोग करने के लिए एक समेकित अनुभव प्रदान करता है, जिससे आप अपने Azure एप्लिकेशन के साथ Confluent Cloud को आसानी से एकीकृत कर सकते हैं।
• Azure Spring Cloud: Azure Spring Cloud के लिए धन्यवाद, Spring Boot माइक्रोसर्विसेज को Azure पर तैनात करना आसान है। Azure Spring Cloud इन्फ्रास्ट्रक्चर चिंताओं को कम करता है, कॉन्फिगरेशन मैनेजमेंट, सर्विस डिस्कवरी, CI/CD एकीकरण, ब्लू-ग्रीन डिप्लॉयमेंट्स, और बहुत कुछ प्रदान करता है। सेवा सभी भारी काम करती है ताकि डेवलपर्स अपने कोड पर ध्यान केंद्रित कर सकें।

कृपया ध्यान दें कि कुछ सेवाओं को स्थानीय रूप से होस्ट किया गया था बस चीजों को सरल रखने के लिए। प्रोडक्शन-ग्रेड डिप्लॉयमेंट्स में आप उन्हें Azure में भी चलाना चाहेंगे। उदाहरण के लिए, आप MQTT कनेक्टर के साथ Kafka Connect क्लस्टर को Azure Kubernetes Service में संचालित कर सकते हैं।

संक्षेप में, यहां एंड-टू-एंड फ्लो है:

• एक स्क्रिप्ट नकली डिवाइस डेटा उत्पन्न करती है जिसे स्थानीय MQTT ब्रोकर को भेजा जाता है।
• यह डेटा MQTT Kafka Connect स्रोत कनेक्टर द्वारा उठाया जाता है और Azure में चल रहे Confluent Cloud Kafka क्लस्टर में एक टॉपिक पर भेजा जाता है।
• इसे Azure Spring Cloud में होस्ट किए गए Spring Boot एप्लिकेशन द्वारा आगे प्रोसेस किया जाता है, जो तब इसे Azure Cache for Redis इंस्टेंस में स्थायी करता है।

व्यावहारिक चीजों से शुरू करने का समय! उससे पहले, सुनिश्चित करें कि आपके पास निम्नलिखित हैं।

पूर्वापेक्षाएँ:

• एक Azure खाता —आप यहां मुफ्त में एक प्राप्त कर सकते हैं
• Azure CLI इंस्टॉल करें
• JDK 11 उदाहरण के लिए OpenJDK
• Maven और Git का एक हालिया संस्करण

इन्फ्रास्ट्रक्चर घटकों को सेट अप करें

RedisTimeSeries मॉड्यूल के साथ आने वाले Azure Cache for Redis (Enterprise Tier) को प्रोविजन करने के लिए दस्तावेज़ीकरण का पालन करें।

Azure Marketplace पर Confluent Cloud क्लस्टर प्रोविजन करें। साथ ही एक Kafka टॉपिक बनाएं (नाम mqtt.device-stats का उपयोग करें) और क्रेडेंशियल्स (API कुंजी और सीक्रेट) बनाएं जिनका उपयोग आप बाद में अपने क्लस्टर से सुरक्षित रूप से कनेक्ट करने के लिए करेंगे।

आप Azure पोर्टल का उपयोग करके Azure Spring Cloud का एक इंस्टेंस प्रोविजन कर सकते हैं या Azure CLI का उपयोग करें:

az spring-cloud create -n <Azure Spring Cloud सेवा का नाम> -g <संसाधन समूह का नाम> -l <स्थान दर्ज करें जैसे southeastasia>

आगे बढ़ने से पहले, GitHub रेपो को क्लोन करना सुनिश्चित करें:

git clone https://github.com/abhirockzz/redis-timeseries-kafka cd redis-timeseries-kafka

स्थानीय सेवाएं सेट करें

घटकों में शामिल हैं:

• Mosquitto MQTT ब्रोकर
• MQTT स्रोत कनेक्टर के साथ Kafka Connect
• डैशबोर्ड में टाइम-सीरीज डेटा ट्रैक करने के लिए Grafana

MQTT ब्रोकर

मैंने Mac पर स्थानीय रूप से mosquitto ब्रोकर इंस्टॉल और शुरू किया।

brew install mosquitto brew services start mosquitto

आप अपने OS के अनुरूप चरणों का पालन कर सकते हैं या इस Docker इमेज का उपयोग करने के लिए स्वतंत्र महसूस कर सकते हैं।

Grafana

मैंने Mac पर स्थानीय रूप से Grafana इंस्टॉल और शुरू किया।

brew install grafana brew services start grafana

आप अपने OS के लिए भी ऐसा ही कर सकते हैं या इस Docker इमेज का उपयोग करने के लिए स्वतंत्र महसूस कर सकते हैं।

docker run -d -p 3000:3000 --name=grafana -e "GF_INSTALL_PLUGINS=redis-datasource" grafana/grafana

Kafka Connect

आपको अभी क्लोन किए गए रेपो में connect-distributed.properties फ़ाइल ढूंढने में सक्षम होना चाहिए। bootstrap.servers, sasl.jaas.config आदि जैसे गुणों के मान बदलें।

सबसे पहले, Apache Kafka को स्थानीय रूप से डाउनलोड और अनज़िप करें।

एक स्थानीय Kafka Connect क्लस्टर शुरू करें:

export KAFKA_INSTALL_DIR=<kafka इंस्टॉलेशन डायरेक्टरी जैसे /home/foo/kafka_2.12-2.5.0> $KAFKA_INSTALL_DIR/bin/connect-distributed.sh connect-distributed.properties

MQTT स्रोत कनेक्टर को मैन्युअल रूप से इंस्टॉल करने के लिए:

• इस लिंक से कनेक्टर/प्लगइन ZIP फ़ाइल डाउनलोड करें, और
• इसे Connect वर्कर के plugin.path कॉन्फिगरेशन गुणों पर सूचीबद्ध निर्देशिकाओं में से एक में निकालें

यदि आप स्थानीय रूप से Confluent Platform का उपयोग कर रहे हैं, तो बस Confluent Hub CLI का उपयोग करें: confluent-hub install confluentinc/kafka-connect-mqtt:latest

MQTT स्रोत कनेक्टर इंस्टेंस बनाएं

mqtt-source-config.json फ़ाइल की जांच करना सुनिश्चित करें। सुनिश्चित करें कि आप kafka.topic के लिए सही टॉपिक नाम दर्ज करते हैं और mqtt.topics को अपरिवर्तित छोड़ देते हैं।

curl -X POST -H 'Content-Type: application/json' https://localhost:8083/connectors -d @mqtt-source-config.json # कनेक्टर स्थिति की जांच करने से पहले एक मिनट प्रतीक्षा करें curl https://localhost:8083/connectors/mqtt-source/status

डिवाइस डेटा प्रोसेसर एप्लिकेशन तैनात करें

आपने अभी जो GitHub रेपो क्लोन किया है, उसमें consumer/src/resources फ़ोल्डर में application.yaml फ़ाइल देखें और निम्न के लिए मान बदलें:

• Azure Cache for Redis होस्ट, पोर्ट और प्राथमिक एक्सेस कुंजी
• Azure पर Confluent Cloud API कुंजी और सीक्रेट

एप्लिकेशन JAR फ़ाइल बनाएं:

cd consumer export JAVA_HOME=<पूर्ण पथ दर्ज करें जैसे /Library/Java/JavaVirtualMachines/zulu-11.jdk/Contents/Home> mvn clean package

एक Azure Spring Cloud एप्लिकेशन बनाएं और JAR फ़ाइल को इसमें तैनात करें:

az spring-cloud app create -n device-data-processor -s <Azure Spring Cloud इंस्टेंस का नाम> -g <संसाधन समूह का नाम> --runtime-version Java_11 az spring-cloud app deploy -n device-data-processor -s <Azure Spring Cloud इंस्टेंस का नाम> -g <संसाधन समूह का नाम> --jar-path target/device-data-processor-0.0.1-SNAPSHOT.jar

नकली डिवाइस डेटा जनरेटर शुरू करें

आप अभी क्लोन किए गए GitHub रेपो में स्क्रिप्ट का उपयोग कर सकते हैं:

./gen-timeseries-data.sh

नोट—यह सब mosquitto_pub CLI कमांड का उपयोग करके डेटा भेजता है।

डेटा device-stats MQTT टॉपिक पर भेजा जाता है (यह नहीं Kafka टॉपिक है)। आप CLI सब्सक्राइबर का उपयोग करके दोबारा जांच सकते हैं:

mosquitto_sub -h localhost -t device-stats

Confluent Cloud पोर्टल में Kafka टॉपिक की जांच करें। आपको Azure Spring Cloud में डिवाइस डेटा प्रोसेसर ऐप के लॉग भी जांचने चाहिए:

az spring-cloud app logs -f -n device-data-processor -s <Azure Spring Cloud इंस्टेंस का नाम> -g <संसाधन समूह का नाम>

Grafana डैशबोर्ड्स का आनंद लें!

Grafana UI को localhost:3000 पर ब्राउज़ करें।

Grafana के लिए Redis Data Source प्लगइन किसी भी Redis डेटाबेस के साथ काम करता है, जिसमें Azure Cache for Redis भी शामिल है। डेटा स्रोत को कॉन्फिगर करने के लिए इस ब्लॉग पोस्ट में निर्देशों का पालन करें।

GitHub रेपो में grafana_dashboards फ़ोल्डर में डैशबोर्ड आयात करें (यदि आपको डैशबोर्ड आयात करने में सहायता की आवश्यकता है तो Grafana दस्तावेज़ीकरण देखें)।

उदाहरण के लिए, यहां एक डैशबोर्ड है जो स्थान 1 में डिवाइस 5 के लिए औसत दबाव (30 सेकंड से अधिक) दिखाता है (TS.MRANGE का उपयोग करता है)।

यहां एक और डैशबोर्ड है जो स्थान 3 में कई डिवाइसेज के लिए अधिकतम तापमान (15 सेकंड से अधिक) दिखाता है (फिर से, TS.MRANGE के लिए धन्यवाद)।

तो, आप कुछ RedisTimeSeries कमांड चलाना चाहते हैं?

redis-cli को चालू करें और Azure Cache for Redis इंस्टेंस से कनेक्ट करें:

redis-cli -h <azure redis होस्टनाम जैसे myredis.southeastasia.redisenterprise.cache.azure.net> -p 10000 -a <azure redis एक्सेस कुंजी> --tls

सरल क्वेरीज़ से शुरू करें:

# स्थान 1 के लिए डिवाइस 5 में दबाव TS.GET pressure:1:5 # स्थान 4 के लिए डिवाइस 5 में तापमान TS.GET temp:4:5

स्थान के अनुसार फ़िल्टर करें और सभी डिवाइसेज के लिए तापमान और दबाव प्राप्त करें:

TS.MGET WITHLABELS FILTER location=3

एक विशिष्ट समय सीमा के भीतर एक या अधिक स्थानों में सभी डिवाइसेज के लिए तापमान और दबाव निकालें:

TS.MRANGE - + WITHLABELS FILTER location=3 TS.MRANGE - + WITHLABELS FILTER location=(3,5)

– + शुरुआत से लेकर नवीनतम टाइमस्टैम्प तक सब कुछ को संदर्भित करता है, लेकिन आप अधिक विशिष्ट हो सकते हैं।

MRANGE वही है जिसकी हमें जरूरत थी! हम किसी स्थान में एक विशिष्ट डिवाइस द्वारा भी फ़िल्टर कर सकते हैं और आगे तापमान या दबाव द्वारा ड्रिल डाउन कर सकते हैं:

TS.MRANGE - + WITHLABELS FILTER location=3 device=2 TS.MRANGE - + WITHLABELS FILTER location=3 metric=temp TS.MRANGE - + WITHLABELS FILTER location=3 device=2 metric=temp

इन सभी को एकत्रीकरण के साथ जोड़ा जा सकता है।

# सभी टेम्प डेटा पॉइंट्स उपयोगी नहीं हैं। हर टेम्प डेटा पॉइंट के बजाय औसत (या अधिकतम) के बारे में क्या? TS.MRANGE - + WITHLABELS AGGREGATION avg 10000 FILTER location=3 metric=temp TS.MRANGE - + WITHLABELS AGGREGATION max 10000 FILTER location=3 metric=temp

एक नियम बनाकर इस एकत्रीकरण को करना और इसे एक अलग टाइम सीरीज में संग्रहीत करना भी संभव है।

एक बार जब आप पूरा कर लें, तो अवांछित लागतों से बचने के लिए संसाधनों को हटाना न भूलें।

संसाधन हटाएं

• Confluent Cloud क्लस्टर को हटाने के लिए दस्तावेज़ीकरण में चरणों का पालन करें—आपको केवल Confluent संगठन को हटाने की आवश्यकता है।
• इसी तरह, आपको Azure Cache for Redis इंस्टेंस को भी हटाना चाहिए।

आपकी स्थानीय मशीन पर:

• Kafka Connect क्लस्टर रोकें
• mosquito ब्रोकर रोकें (जैसे brew services stop mosquito)
• Grafana सेवा रोकें (जैसे brew services stop grafana)

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

अतिरिक्त विचार

RedisTimeSeries को ऑप्टिमाइज़ करना

• रिटेंशन पॉलिसी: इस पर विचार करें क्योंकि आपके टाइम-सीरीज डेटा पॉइंट्स डिफ़ॉल्ट रूप से नहीं ट्रिम या डिलीट होते हैं।
• डाउन-सैंपलिंग और एकत्रीकरण नियम: आप डेटा को हमेशा के लिए स्टोर नहीं करना चाहते, है ना? इसका ख्याल रखने के लिए उचित नियम कॉन्फिगर करना सुनिश्चित करें (जैसे TS.CREATERULE temp:1:2 temp:avg:30 AGGREGATION avg 30000)।
• डुप्लिकेट डेटा पॉलिसी: आप डुप्लिकेट सैंपल्स को कैसे संभालना चाहेंगे? सुनिश्चित करें कि डिफ़ॉल्ट पॉलिसी (BLOCK) वास्तव में वही है जो आपको चाहिए। यदि नहीं, तो अन्य विकल्पों पर विचार करें।

यह एक विस्तृत सूची नहीं है। अन्य कॉन्फिगरेशन विकल्पों के लिए, कृपया RedisTimeSeries दस्तावेज़ीकरण देखें

लंबी अवधि के डेटा प्रतिधारण के बारे में क्या?

डेटा कीमती है, टाइम सीरीज सहित! आप इसे आगे प्रोसेस करना चाह सकते हैं (जैसे अंतर्दृष्टि निकालने के लिए मशीन लर्निंग चलाना, प्रेडिक्टिव मेंटेनेंस, आदि)। इसके लिए संभव होने के लिए, आपको इस डेटा को लंबे समय तक बनाए रखने की आवश्यकता होगी, और इसके लिए लागत-प्रभावी और कुशल होने के लिए, आप Azure Data Lake Storage Gen2 (ADLS Gen2) जैसी स्केलेबल ऑब्जेक्ट स्टोरेज सेवा का उपयोग करना चाहेंगे।

उसके लिए एक कनेक्टर है! आप पूरी तरह से प्रबंधित Azure Data Lake Storage Gen2 सिंक कनेक्टर का उपयोग करके मौजूदा डेटा पाइपलाइन को बढ़ा सकते हैं, Confluent Cloud के लिए डेटा को प्रोसेस और ADLS में स्टोर करने के लिए, और फिर Azure Synapse Analytics या Azure Databricks का उपयोग करके मशीन लर्निंग चलाएं।

स्केलेबिलिटी

आपके टाइम-सीरीज डेटा वॉल्यूम केवल एक ही दिशा में जा सकते हैं—ऊपर! आपके समाधान का स्केलेबल होना महत्वपूर्ण है:

• कोर इन्फ्रास्ट्रक्चर: प्रबंधित सेवाएं टीमों को समाधान पर ध्यान केंद्रित करने की अनुमति देती हैं बजाय इन्फ्रास्ट्रक्चर को सेट अप और बनाए रखने के, खासकर जब Redis और Kafka जैसे जटिल वितरित सिस्टम की बात आती है।
• Kafka Connect: जहां तक डेटा पाइपलाइन का संबंध है, आप अच्छे हाथों में हैं क्योंकि Kafka Connect प्लेटफॉर्म स्वाभाविक रूप से स्टेटलेस और क्षैतिज रूप से स्केलेबल है। आपके पास इस बात के बहुत सारे विकल्प हैं कि आप अपने Kafka Connect वर्कर क्लस्टर्स को कैसे आर्किटेक्ट और आकार देना चाहते हैं।
• कस्टम एप्लिकेशन: जैसा कि इस समाधान में मामला था, हमने Kafka टॉपिक्स में डेटा प्रोसेस करने के लिए एक कस्टम एप्लिकेशन बनाया। सौभाग्य से, वही स्केलेबिलिटी विशेषताएं उन पर भी लागू होती हैं। क्षैतिज स्केल के संदर्भ में, यह केवल आपके द्वारा रखे गए Kafka टॉपिक पार्टीशन की संख्या तक सीमित है।

एकीकरण: यह केवल Grafana नहीं है! RedisTimeSeries Prometheus और Telegraf के साथ भी एकीकृत होता है। हालांकि, इस ब्लॉग पोस्ट के लिखे जाने के समय कोई Kafka कनेक्टर नहीं है—यह एक शानदार ऐड-ऑन होगा!

निष्कर्ष

निश्चित रूप से, आप Redis का उपयोग (लगभग) हर चीज के लिए कर सकते हैं, जिसमें टाइम-सीरीज वर्कलोड भी शामिल हैं! टाइम-सीरीज डेटा स्रोतों से लेकर Redis और उससे आगे तक, डेटा पाइपलाइन और एकीकरण के लिए एंड-टू-एंड आर्किटेक्चर के बारे में सोचना सुनिश्चित करें।