🚣🏼 🏉 🤹🏻 PostgreSQL: PipelineDB - वास्तविक समय के समग्र प्रश्न 🎐 👈🏽 🎏

क्या कभी आपको पिछले महीने के लिए डेटाबेस में डेटा के आधार पर किसी चीज़ की मात्रा की गणना करने के लिए कहा गया है, जिसके परिणामस्वरूप कुछ मूल्यों को समूहबद्ध किया गया है और इसे दिन / घंटे के हिसाब से पूरा किया जा रहा है?
यदि हाँ - तो आप पहले से ही कल्पना करते हैं कि आपको कुछ इस तरह से लिखना है, केवल बदतर

SELECT hour(datetime), somename, count(*), sum(somemetric) from table where datetime > :monthAgo group by 1, 2 order by 1 desc, 2

समय-समय पर, इस तरह के अनुरोधों की एक विस्तृत विविधता दिखाई देने लगती है, और यदि आप एक बार सहन करते हैं और मदद करते हैं, तो भविष्य में अपील आएगी।

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

लेकिन क्या होगा अगर मैं कहता हूं कि सीधे PostgreSQL में आप एक दृश्य बना सकते हैं कि मक्खी सीधे ऊपर के रूप में एक सीधे समान क्वेरी में नए आने वाले डेटा पर विचार करेगी?

तो - यह PipelineDB एक्सटेंशन कर सकता है

उनकी साइट से डेमो कि यह कैसे काम करता है

PipelineDB पहले एक अलग परियोजना थी, लेकिन अब PG 10.1 और उच्चतर के लिए एक एक्सटेंशन के रूप में उपलब्ध है।

और यद्यपि प्रदान किए गए अवसर लंबे समय से अन्य उत्पादों में मौजूद हैं, जो विशेष रूप से वास्तविक समय के मैट्रिक्स को इकट्ठा करने के लिए डिज़ाइन किए गए हैं, PipelineDB का एक महत्वपूर्ण प्लस है: डेवलपर्स के लिए एक कम प्रवेश सीमा जो पहले से ही SQL जानते हैं)।

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

और इसलिए, सामान्य तौर पर, उनके पास अच्छे दस्तावेज हैं, लेकिन मैं अपने अनुभव को साझा करना चाहता हूं कि इस व्यवसाय को व्यवहार में कैसे आजमाया जाए और परिणामों को ग्राफ्टाना में लाया जाए।

स्थानीय मशीन को खराब न करने के लिए, मैं सब कुछ डॉकटर में तैनात करता हूं।
उपयोग की गई छवियाँ: postgres:latest , grafana/grafana

Postgres पर PipelineDB स्थापित करें

पोस्टग्रेज के साथ एक मशीन पर, क्रमिक रूप से निष्पादित करें:

apt update
apt install curl
curl -s http://download.pipelinedb.com/apt.sh | bash
apt install pipelinedb-postgresql-11
cd /var/lib/postgresql/data
किसी भी संपादक में postgresql.conf फ़ाइल खोलें
shared_preload_libraries कुंजी को ढूंढें, अनप्लग करें और pipelinedb मान सेट करें
मुख्य max_worker_processes 128 पर सेट (अनुशंसा डॉक)
रिबूट सर्वर

PipelineDB में एक स्ट्रीम और दृश्य बनाना

रीबूट पीजी के बाद - लॉग देखें ताकि ऐसा कुछ हो

वह डेटाबेस जिसमें हम काम करेंगे: CREATE DATABASE testpipe;
एक एक्सटेंशन बनाना: CREATE EXTENSION pipelinedb;
अब सबसे दिलचस्प बात एक धारा बना रही है। यह इसमें है कि आपको आगे की प्रक्रिया के लिए डेटा जोड़ने की आवश्यकता है:
```
 CREATE FOREIGN TABLE flow_stream ( dtmsk timestamp without time zone, action text, duration smallint ) SERVER pipelinedb; 
```
वास्तव में, यह एक साधारण तालिका बनाने के समान है, आप इस धारा से केवल एक साधारण select साथ डेटा प्राप्त नहीं कर सकते - आपको एक दृश्य की आवश्यकता है
वास्तव में इसे कैसे बनाया जाए:
```
 CREATE VIEW viewflow WITH (ttl = '3 month', ttl_column = 'm') AS select minute(dtmsk) m, action, count(*), avg(duration)::smallint, min(duration), max(duration) from flow_stream group by 1, 2; 
```
उन्हें कंटीन्यूअस व्यूज़ कहा जाता है और उन्हें भौतिक बनाने के लिए डिफ़ॉल्ट किया जाता है, अर्थात राज्य के संरक्षण के साथ।

WITH अतिरिक्त पैरामीटर पास करता है।

मेरे मामले में, ttl = '3 month' अर्थ है कि आपको केवल पिछले 3 महीनों के लिए डेटा संग्रहीत करने की आवश्यकता है, और कॉलम M से दिनांक / समय ले सकते हैं M पृष्ठभूमि reaper प्रक्रिया अप्रचलित डेटा की तलाश करती है और इसे हटा देती है।

जो लोग जानते नहीं हैं, उनके लिए minute फ़ंक्शन सेकंड के बिना दिनांक / समय देता है। इस प्रकार, एक मिनट में होने वाली सभी घटनाओं के एकत्रीकरण के परिणामस्वरूप एक ही समय होगा।
ऐसा दृश्य लगभग एक तालिका है, क्योंकि यदि बहुत सारा डेटा संग्रहीत किया जाता है, तो नमूने के लिए तारीख तक सूचकांक उपयोगी होगा
```
 create index on viewflow (m desc, action); 
```

PipelineDB का उपयोग करना

याद रखें: स्ट्रीम में डेटा डालें, और सदस्यता लेने वाले दृश्य से पढ़ें

 insert into flow_stream VALUES (now(), 'act1', 21); insert into flow_stream VALUES (now(), 'act2', 33); select * from viewflow order by m desc, action limit 4; select now()

मैं मैन्युअल रूप से अनुरोध निष्पादित करता हूं

पहले मैं देखता हूं कि 46 वें मिनट में डेटा कैसे बदलता है
जैसे ही 47 वाँ आता है, पहले वाला अपडेट करना बंद कर देता है और चालू मिनट टिकना शुरू कर देता है।

यदि आप क्वेरी योजना पर ध्यान देते हैं, तो आप डेटा के साथ मूल तालिका देख सकते हैं

मैं इसे जाने की सलाह देता हूं और यह पता लगाता हूं कि वास्तव में आपका डेटा कैसे संग्रहीत है

C # ईवेंट जेनरेटर

 using Npgsql; using System; using System.Threading; namespace PipelineDbLogGenerator { class Program { private static Random _rnd = new Random(); private static string[] _actions = new string[] { "foo", "bar", "yep", "goal", "ano" }; static void Main(string[] args) { var connString = "Host=localhost;port=5432;Username=postgres;Database=testpipe"; using (var conn = new NpgsqlConnection(connString)) { conn.Open(); while (true) { var dt = DateTime.UtcNow; using (var cmd = new NpgsqlCommand()) { var act = GetAction(); cmd.Connection = conn; cmd.CommandText = "INSERT INTO flow_stream VALUES (@dtmsk, @action, @duration)"; cmd.Parameters.AddWithValue("dtmsk", dt); cmd.Parameters.AddWithValue("action", act); cmd.Parameters.AddWithValue("duration", GetDuration(act)); var res = cmd.ExecuteNonQuery(); Console.WriteLine($"{res} {dt}"); } Thread.Sleep(_rnd.Next(50, 230)); } } } private static int GetDuration(string act) { var c = 0; for (int i = 0; i < act.Length; i++) { c += act[i]; } return _rnd.Next(c); } private static string GetAction() { return _actions[_rnd.Next(_actions.Length)]; } } }

गुराफना में निष्कर्ष

पोस्टग्रेज से डेटा प्राप्त करने के लिए, आपको उचित डेटा स्रोत जोड़ना होगा:

एक नया डैशबोर्ड बनाएं और उसमें ग्राफ़ के प्रकार का एक पैनल जोड़ें, और उसके बाद आपको पैनल के संपादन में जाने की आवश्यकता है:

अगला - एक डेटा स्रोत का चयन करें, sql- क्वेरी लेखन मोड पर जाएं और इसे दर्ज करें:

 select m as time, -- Grafana   time count, action from viewflow where $__timeFilter(m) --  ,     ,   col between :startdate and :enddate order by m desc, action;

और फिर आपको एक सामान्य शेड्यूल मिलता है, निश्चित रूप से, यदि आपने इवेंट जनरेटर शुरू किया

FYI करें: एक इंडेक्स होना बहुत महत्वपूर्ण हो सकता है। यद्यपि इसका उपयोग परिणामी तालिका की मात्रा पर निर्भर करता है। यदि आप कम समय में छोटी संख्या में पंक्तियों को संग्रहीत करने की योजना बनाते हैं, तो यह बहुत आसानी से पता लगा सकता है कि seq स्कैन सस्ता होगा, और सूचकांक केवल अतिरिक्त जोड़ देगा। मूल्यों को अपडेट करते समय लोड करें

एकाधिक विचारों को एक स्ट्रीम में सब्सक्राइब किया जा सकता है।

मान लीजिए मैं देखना चाहता हूं कि प्रतिशत के आधार पर कितने एपी तरीके हैं

 CREATE VIEW viewflow_per WITH (ttl = '3 d', ttl_column = 'm') AS select minute(dtmsk) m, action, percentile_cont(0.50) WITHIN GROUP (ORDER BY duration)::smallint p50, percentile_cont(0.95) WITHIN GROUP (ORDER BY duration)::smallint p95, percentile_cont(0.99) WITHIN GROUP (ORDER BY duration)::smallint p99 from flow_stream group by 1, 2; create index on viewflow_per (m desc);

मैं ग्राफ्टाना के साथ एक ही चाल है और मिलता है:

कुल मिलाकर

सामान्य तौर पर, बात काम कर रही है, यह शिकायतों के बिना, अच्छा व्यवहार करती है। हालांकि डॉकटर के तहत, संग्रह (2.3 जीबी) में उनके डेमो डेटाबेस को डाउनलोड करना थोड़ा लंबा हो गया।

मैं ध्यान देना चाहता हूं - मैंने तनाव परीक्षण नहीं किया।

आधिकारिक दस्तावेज

दिलचस्प हो सकता है

Apache Kafka से स्ट्रीम में डेटा डाउनलोड करने के लिए समर्थन
Amazon Kinesis के समान
आप केवल डेटा परिवर्तन (भंडारण के बिना) के लिए विचार बना सकते हैं
PipelineDB क्लस्टर - एक व्यावसायिक संस्करण है। इसमें, आप शार्क के अनुसार विचार वितरित कर सकते हैं। क्लस्टर निर्णय डॉक में अधिक जानकारी