वास्तविक समय सेवा के उदाहरण का उपयोग करके Q और KDB+ भाषा की विशेषताएं

आप मेरे पिछले लेख में पढ़ सकते हैं कि केडीबी+ बेस, क्यू प्रोग्रामिंग भाषा क्या हैं, उनकी ताकत और कमजोरियां क्या हैं। लेख और संक्षेप में परिचय में। लेख में, हम क्यू पर एक सेवा लागू करेंगे जो आने वाली डेटा स्ट्रीम को संसाधित करेगी और "वास्तविक समय" मोड में हर मिनट विभिन्न एकत्रीकरण कार्यों की गणना करेगी (यानी, इसमें डेटा के अगले भाग से पहले सब कुछ की गणना करने का समय होगा)। Q की मुख्य विशेषता यह है कि यह एक वेक्टर भाषा है जो आपको एकल वस्तुओं के साथ नहीं, बल्कि उनके सरणियों, सरणियों के सरणियों और अन्य जटिल वस्तुओं के साथ काम करने की अनुमति देती है। Q और उसके रिश्तेदार K, J, APL जैसी भाषाएँ अपनी संक्षिप्तता के लिए प्रसिद्ध हैं। अक्सर, एक प्रोग्राम जो जावा जैसी परिचित भाषा में कोड की कई स्क्रीन लेता है, उन पर कुछ पंक्तियों में लिखा जा सकता है। इस लेख में मैं यही प्रदर्शित करना चाहता हूं।

परिचय

केडीबी+ एक स्तंभकार डेटाबेस है जो बहुत बड़ी मात्रा में डेटा पर केंद्रित होता है, जिसे एक विशिष्ट तरीके से (मुख्य रूप से समय के अनुसार) ऑर्डर किया जाता है। इसका उपयोग मुख्य रूप से वित्तीय संस्थानों - बैंकों, निवेश कोष, बीमा कंपनियों में किया जाता है। Q भाषा KDB+ की आंतरिक भाषा है जो आपको इस डेटा के साथ प्रभावी ढंग से काम करने की अनुमति देती है। क्यू विचारधारा संक्षिप्तता और दक्षता है, जबकि स्पष्टता का त्याग किया जाता है। यह इस तथ्य से उचित है कि वेक्टर भाषा को किसी भी मामले में समझना मुश्किल होगा, और रिकॉर्डिंग की संक्षिप्तता और समृद्धि आपको एक स्क्रीन पर प्रोग्राम का बहुत बड़ा हिस्सा देखने की अनुमति देती है, जो अंततः इसे समझना आसान बनाती है।

इस लेख में हम Q में एक पूर्ण कार्यक्रम लागू करते हैं और हो सकता है कि आप इसे आज़माना चाहें। ऐसा करने के लिए, आपको वास्तविक Q की आवश्यकता होगी। आप kx कंपनी की वेबसाइट पर मुफ्त 32-बिट संस्करण डाउनलोड कर सकते हैं - www.kx.com. वहां, यदि आप रुचि रखते हैं, तो आपको क्यू, पुस्तक पर संदर्भ जानकारी मिलेगी क्यू नश्वर लोगों के लिए और इस विषय पर विभिन्न लेख।

समस्या का निरूपण

एक स्रोत है जो हर 25 मिलीसेकंड में डेटा के साथ एक तालिका भेजता है। चूँकि KDB+ का उपयोग मुख्य रूप से वित्त में किया जाता है, हम मान लेंगे कि यह लेनदेन (व्यापार) की एक तालिका है, जिसमें निम्नलिखित कॉलम हैं: समय (मिलीसेकंड में समय), सिम (स्टॉक एक्सचेंज पर कंपनी का पदनाम - आईबीएम, AAPL,…), कीमत (वह कीमत जिस पर शेयर खरीदे गए थे), आकार (लेन-देन का आकार)। 25 मिलीसेकंड का अंतराल मनमाना है, न बहुत छोटा और न बहुत लंबा। इसकी उपस्थिति का मतलब है कि डेटा पहले से ही बफर्ड सेवा में आता है। वर्तमान लोड के आधार पर गतिशील बफरिंग सहित सेवा पक्ष पर बफरिंग लागू करना आसान होगा, लेकिन सादगी के लिए, हम एक निश्चित अंतराल पर ध्यान केंद्रित करेंगे।

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

हम यह भी मानेंगे कि आने वाली डेटा स्ट्रीम समयबद्ध है। इससे हमें अंतिम समय में ही काम करने का मौका मिलेगा।' व्यवहार में, कुछ अपडेट देर से होने की स्थिति में वर्तमान और पिछले मिनटों के साथ काम करने में सक्षम होना ही पर्याप्त है। सरलता के लिए, हम इस मामले पर विचार नहीं करेंगे।

एकत्रीकरण कार्य

आवश्यक एकत्रीकरण कार्य नीचे सूचीबद्ध हैं। सेवा पर भार बढ़ाने के लिए मैंने यथासंभव उनमें से कई को लिया:

• उच्च - अधिकतम मूल्य - प्रति मिनट अधिकतम मूल्य।
• कम - न्यूनतम कीमत - प्रति मिनट न्यूनतम कीमत।
• प्रथम मूल्य - प्रथम मूल्य - प्रथम मूल्य प्रति मिनट।
• अंतिम मूल्य - अंतिम मूल्य - अंतिम मूल्य प्रति मिनट।
• पहला आकार - पहला आकार - प्रति मिनट पहला व्यापार आकार।
• अंतिम आकार - अंतिम आकार - एक मिनट में अंतिम व्यापार आकार।
• numTrades - गिनती i - प्रति मिनट ट्रेडों की संख्या।
• मात्रा - योग आकार - प्रति मिनट व्यापार आकार का योग।
• पीवॉल्यूम - कुल मूल्य - प्रति मिनट कीमतों का योग, औसत मूल्य के लिए आवश्यक।
• - कुल टर्नओवर मूल्य*आकार - प्रति मिनट लेनदेन की कुल मात्रा।
• औसत मूल्य - pvolume%numTrades - प्रति मिनट औसत मूल्य।
• औसत आकार - मात्रा% संख्या व्यापार - प्रति मिनट औसत व्यापार आकार।
• vwap - टर्नओवर%वॉल्यूम - लेनदेन आकार द्वारा भारित प्रति मिनट औसत मूल्य।
• कमवॉल्यूम - योग मात्रा - पूरे समय में लेनदेन का संचित आकार।

आइए तुरंत एक गैर-स्पष्ट बिंदु पर चर्चा करें - इन कॉलमों को पहली बार और प्रत्येक बाद के मिनट के लिए कैसे आरंभ किया जाए। फर्स्टप्राइस प्रकार के कुछ कॉलमों को हर बार शून्य करने के लिए प्रारंभ किया जाना चाहिए; उनका मान अपरिभाषित है। अन्य वॉल्यूम प्रकारों को हमेशा 0 पर सेट किया जाना चाहिए। ऐसे कॉलम भी हैं जिनके लिए एक संयुक्त दृष्टिकोण की आवश्यकता होती है - उदाहरण के लिए, कम वॉल्यूम को पिछले मिनट से कॉपी किया जाना चाहिए, और पहले वाले के लिए 0 पर सेट किया जाना चाहिए। आइए शब्दकोश डेटा का उपयोग करके इन सभी मापदंडों को सेट करें प्रकार (रिकॉर्ड के अनुरूप):

// list ! list – создать словарь, 0n – float null, 0N – long null, `sym – тип символ, `sym1`sym2 – список символов
initWith:`sym`time`high`low`firstPrice`lastPrice`firstSize`lastSize`numTrades`volume`pvolume`turnover`avgPrice`avgSize`vwap`cumVolume!(`;00:00;0n;0n;0n;0n;0N;0N;0;0;0.0;0.0;0n;0n;0n;0);
aggCols:reverse key[initWith] except `sym`time; // список всех вычисляемых колонок, reverse объяснен ниже

मैंने सुविधा के लिए शब्दकोश में सिम और समय जोड़ा है, अब initWith अंतिम एकत्रित तालिका से एक तैयार लाइन है, जहां यह सही सिम और समय सेट करने के लिए बनी हुई है। आप इसका उपयोग किसी तालिका में नई पंक्तियाँ जोड़ने के लिए कर सकते हैं।

एकत्रीकरण फ़ंक्शन बनाते समय हमें aggCols की आवश्यकता होगी। जिस क्रम में क्यू में अभिव्यक्तियों का मूल्यांकन किया जाता है (दाएं से बाएं) के कारण सूची उलटी होनी चाहिए। लक्ष्य यह सुनिश्चित करना है कि गणना उच्च से कम वॉल्यूम तक जाए, क्योंकि कुछ कॉलम पिछले वाले पर निर्भर करते हैं।

जिन कॉलमों को पिछले वाले से एक नए मिनट में कॉपी करने की आवश्यकता होती है, सुविधा के लिए सिम कॉलम जोड़ा जाता है:

rollColumns:`sym`cumVolume;

आइए अब कॉलमों को इस अनुसार समूहों में विभाजित करें कि उन्हें कैसे अद्यतन किया जाना चाहिए। तीन प्रकारों को प्रतिष्ठित किया जा सकता है:

1. संचायक (वॉल्यूम, टर्नओवर,..) - हमें आने वाले मूल्य को पिछले मूल्य में जोड़ना होगा।
2. एक विशेष बिंदु (उच्च, निम्न, ..) के साथ - मिनट में पहला मान आने वाले डेटा से लिया जाता है, बाकी की गणना फ़ंक्शन का उपयोग करके की जाती है।
3. आराम। हमेशा एक फ़ंक्शन का उपयोग करके गणना की जाती है।

आइए इन वर्गों के लिए चर परिभाषित करें:

accumulatorCols:`numTrades`volume`pvolume`turnover;
specialCols:`high`low`firstPrice`firstSize;

गणना क्रम

हम समग्र तालिका को दो चरणों में अद्यतन करेंगे। दक्षता के लिए, हम पहले आने वाली तालिका को छोटा करते हैं ताकि प्रत्येक वर्ण और मिनट के लिए केवल एक पंक्ति हो। तथ्य यह है कि हमारे सभी कार्य वृद्धिशील और सहयोगी हैं, यह गारंटी देता है कि इस अतिरिक्त कदम का परिणाम नहीं बदलेगा। आप चयन का उपयोग करके तालिका को छोटा कर सकते हैं:

select high:max price, low:min price … by sym,time.minute from table

इस पद्धति का एक नुकसान है - परिकलित स्तंभों का सेट पूर्वनिर्धारित है। सौभाग्य से, क्यू में, चयन को एक फ़ंक्शन के रूप में भी कार्यान्वित किया जाता है जहां आप गतिशील रूप से बनाए गए तर्कों को प्रतिस्थापित कर सकते हैं:

?[table;whereClause;byClause;selectClause]

मैं तर्कों के प्रारूप का विस्तार से वर्णन नहीं करूंगा; हमारे मामले में, केवल द्वारा और चयनित अभिव्यक्तियां गैर-तुच्छ होंगी और उन्हें प्रपत्र कॉलम!अभिव्यक्तियों के शब्दकोश होने चाहिए। इस प्रकार, सिकुड़न फ़ंक्शन को निम्नानुसार परिभाषित किया जा सकता है:

selExpression:`high`low`firstPrice`lastPrice`firstSize`lastSize`numTrades`volume`pvolume`turnover!parse each ("max price";"min price";"first price";"last price";"first size";"last size";"count i";"sum size";"sum price";"sum price*size"); // each это функция map в Q для одного списка
preprocess:?[;();`sym`time!`sym`time.minute;selExpression];

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

दूसरा चरण समग्र तालिका को अद्यतन करना है। आइए सबसे पहले एल्गोरिदम को स्यूडोकोड में लिखें:

for each sym in inputTable
  idx: row index in agg table for sym+currentTime;
  aggTable[idx;`high]: aggTable[idx;`high] | inputTable[sym;`high];
  aggTable[idx;`volume]: aggTable[idx;`volume] + inputTable[sym;`volume];
  …

क्यू में, लूप के बजाय मैप/रिड्यूस फ़ंक्शंस का उपयोग करना आम है। लेकिन चूँकि Q एक सदिश भाषा है और हम एक ही बार में सभी प्रतीकों पर सभी परिचालनों को आसानी से लागू कर सकते हैं, तो पहले सन्निकटन में हम बिना किसी लूप के सभी प्रतीकों पर एक साथ संचालन कर सकते हैं:

idx:calcIdx inputTable;
row:aggTable idx;
aggTable[idx;`high]: row[`high] | inputTable`high;
aggTable[idx;`volume]: row[`volume] + inputTable`volume;
…

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

idx:calcIdx inputTable;
rows:aggTable idx;
// .[target;(idx0;idx1;..);function;argument] ~ target[idx 0;idx 1;…]: function[target[idx 0;idx 1;…];argument], в нашем случае функция – это присваивание
.[aggTable;(idx;aggCols);:;flip (row[`high] | inputTable`high;row[`volume] + inputTable`volume;…)];

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

.[aggTable;;:;]'[(idx;)each aggCols; (row[`high] | inputTable`high;row[`volume] + inputTable`volume;…)];

हम फिर से फ़ंक्शन प्रक्षेपण का उपयोग करते हैं। यह भी ध्यान दें कि Q में, सूची बनाना भी एक फ़ंक्शन है और हम सूचियों की सूची प्राप्त करने के लिए प्रत्येक (मैप) फ़ंक्शन का उपयोग करके इसे कॉल कर सकते हैं।

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

aggExpression:`high`low`firstPrice`lastPrice`firstSize`lastSize`avgPrice`avgSize`vwap`cumVolume!
 ("row[`high]|inp`high";"row[`low]&inp`low";"row`firstPrice";"inp`lastPrice";"row`firstSize";"inp`lastSize";"pvolume%numTrades";"volume%numTrades";"turnover%volume";"row[`cumVolume]+inp`volume");

कुछ कॉलम विशेष हैं; उनके पहले मान की गणना फ़ंक्शन द्वारा नहीं की जानी चाहिए। हम पंक्ति[`numTrades] कॉलम द्वारा यह निर्धारित कर सकते हैं कि यह पहला है - यदि इसमें 0 है, तो मान पहला है। Q का एक चयन फ़ंक्शन है - ?[बूलियन सूची;सूची1;सूची2] - जो पहले तर्क में स्थिति के आधार पर सूची 1 या 2 से एक मान का चयन करता है:

// high -> ?[isFirst;inp`high;row[`high]|inp`high]
// @ - тоже обобщенное присваивание для случая когда индекс неглубокий
@[`aggExpression;specialCols;{[x;y]"?[isFirst;inp`",y,";",x,"]"};string specialCols];

यहां मैंने अपने फ़ंक्शन (घुंघराले ब्रेसिज़ में एक अभिव्यक्ति) के साथ एक सामान्यीकृत असाइनमेंट बुलाया। यह वर्तमान मान (पहला तर्क) और एक अतिरिक्त तर्क प्राप्त करता है, जिसे मैं चौथे पैरामीटर में पास करता हूं।

आइए बैटरी स्पीकर अलग से जोड़ें, क्योंकि उनके लिए फ़ंक्शन समान है:

// volume -> row[`volume]+inp`volume
aggExpression[accumulatorCols]:{"row[`",x,"]+inp`",x } each string accumulatorCols;

यह Q मानकों द्वारा एक सामान्य असाइनमेंट है, लेकिन मैं एक ही बार में मानों की एक सूची निर्दिष्ट कर रहा हूं। अंत में, आइए मुख्य फ़ंक्शन बनाएं:

// ":",/:aggExprs ~ map[{":",x};aggExpr] => ":row[`high]|inp`high" присвоим вычисленное значение переменной, потому что некоторые колонки зависят от уже вычисленных значений
// string[cols],'exprs ~ map[,;string[cols];exprs] => "high:row[`high]|inp`high" завершим создание присваивания. ,’ расшифровывается как map[concat]
// ";" sv exprs – String from Vector (sv), соединяет список строк вставляя “;” посредине
updateAgg:value "{[aggTable;idx;inp] row:aggTable idx; isFirst_0=row`numTrades; .[aggTable;;:;]'[(idx;)each aggCols;(",(";"sv string[aggCols],'":",/:aggExpression aggCols),")]}";

इस अभिव्यक्ति के साथ, मैं गतिशील रूप से एक स्ट्रिंग से एक फ़ंक्शन बनाता हूं जिसमें ऊपर दी गई अभिव्यक्ति शामिल है। परिणाम इस तरह दिखेगा:

{[aggTable;idx;inp] rows:aggTable idx; isFirst_0=row`numTrades; .[aggTable;;:;]'[(idx;)each aggCols ;(cumVolume:row[`cumVolume]+inp`cumVolume;… ; high:?[isFirst;inp`high;row[`high]|inp`high])]}

स्तंभ मूल्यांकन क्रम उलटा है क्योंकि Q में मूल्यांकन क्रम दाएँ से बाएँ है।

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

अंतिम चरण

हमारे पास प्रीप्रोसेस और अपडेटएजीजी फ़ंक्शन हैं जो सभी काम करते हैं। लेकिन मिनटों के माध्यम से सही संक्रमण सुनिश्चित करना और एकत्रीकरण के लिए सूचकांक की गणना करना अभी भी आवश्यक है। सबसे पहले, आइए init फ़ंक्शन को परिभाषित करें:

init:{
  tradeAgg:: 0#enlist[initWith]; // создаем пустую типизированную таблицу, enlist превращает словарь в таблицу, а 0# означает взять 0 элементов из нее
  currTime::00:00; // начнем с 0, :: означает, что присваивание в глобальную переменную
  currSyms::`u#`symbol$(); // `u# - превращает список в дерево, для ускорения поиска элементов
  offset::0; // индекс в tradeAgg, где начинается текущая минута 
  rollCache:: `sym xkey update `u#sym from rollColumns#tradeAgg; // кэш для последних значений roll колонок, таблица с ключом sym
 }

हम रोल फ़ंक्शन को भी परिभाषित करेंगे, जो वर्तमान मिनट को बदल देगा:

roll:{[tm]
  if[currTime>tm; :init[]]; // если перевалили за полночь, то просто вызовем init
  rollCache,::offset _ rollColumns#tradeAgg; // обновим кэш – взять roll колонки из aggTable, обрезать, вставить в rollCache
  offset::count tradeAgg;
  currSyms::`u#`$();
 }

नए अक्षर जोड़ने के लिए हमें एक फ़ंक्शन की आवश्यकता होगी:

addSyms:{[syms]
  currSyms,::syms; // добавим в список известных
  // добавим в таблицу sym, time и rollColumns воспользовавшись обобщенным присваиванием.
  // Функция ^ подставляет значения по умолчанию для roll колонок, если символа нет в кэше. value flip table возвращает список колонок в таблице.
  `tradeAgg upsert @[count[syms]#enlist initWith;`sym`time,cols rc;:;(syms;currTime), (initWith cols rc)^value flip rc:rollCache ([] sym: syms)];
 }

और अंत में, अपडेट फ़ंक्शन (क्यू सेवाओं के लिए इस फ़ंक्शन का पारंपरिक नाम), जिसे क्लाइंट द्वारा डेटा जोड़ने के लिए कहा जाता है:

upd:{[tblName;data] // tblName нам не нужно, но обычно сервис обрабатывает несколько таблиц 
  tm:exec distinct time from data:() xkey preprocess data; // preprocess & calc time
  updMinute[data] each tm; // добавим данные для каждой минуты
};
updMinute:{[data;tm]
  if[tm<>currTime; roll tm; currTime::tm]; // поменяем минуту, если необходимо
  data:select from data where time=tm; // фильтрация
  if[count msyms:syms where not (syms:data`sym)in currSyms; addSyms msyms]; // новые символы
  updateAgg[`tradeAgg;offset+currSyms?syms;data]; // обновим агрегированную таблицу. Функция ? ищет индекс элементов списка справа в списке слева.
 };

बस इतना ही। यहाँ हमारी सेवा का पूरा कोड है, जैसा कि वादा किया गया था, बस कुछ पंक्तियाँ:

initWith:`sym`time`high`low`firstPrice`lastPrice`firstSize`lastSize`numTrades`volume`pvolume`turnover`avgPrice`avgSize`vwap`cumVolume!(`;00:00;0n;0n;0n;0n;0N;0N;0;0;0.0;0.0;0n;0n;0n;0);
aggCols:reverse key[initWith] except `sym`time;
rollColumns:`sym`cumVolume;

accumulatorCols:`numTrades`volume`pvolume`turnover;
specialCols:`high`low`firstPrice`firstSize;

selExpression:`high`low`firstPrice`lastPrice`firstSize`lastSize`numTrades`volume`pvolume`turnover!parse each ("max price";"min price";"first price";"last price";"first size";"last size";"count i";"sum size";"sum price";"sum price*size");
preprocess:?[;();`sym`time!`sym`time.minute;selExpression];

aggExpression:`high`low`firstPrice`lastPrice`firstSize`lastSize`avgPrice`avgSize`vwap`cumVolume!("row[`high]|inp`high";"row[`low]&inp`low";"row`firstPrice";"inp`lastPrice";"row`firstSize";"inp`lastSize";"pvolume%numTrades";"volume%numTrades";"turnover%volume";"row[`cumVolume]+inp`volume");
@[`aggExpression;specialCols;{"?[isFirst;inp`",y,";",x,"]"};string specialCols];
aggExpression[accumulatorCols]:{"row[`",x,"]+inp`",x } each string accumulatorCols;
updateAgg:value "{[aggTable;idx;inp] row:aggTable idx; isFirst_0=row`numTrades; .[aggTable;;:;]'[(idx;)each aggCols;(",(";"sv string[aggCols],'":",/:aggExpression aggCols),")]}"; / '

init:{
  tradeAgg::0#enlist[initWith];
  currTime::00:00;
  currSyms::`u#`symbol$();
  offset::0;
  rollCache:: `sym xkey update `u#sym from rollColumns#tradeAgg;
 };
roll:{[tm]
  if[currTime>tm; :init[]];
  rollCache,::offset _ rollColumns#tradeAgg;
  offset::count tradeAgg;
  currSyms::`u#`$();
 };
addSyms:{[syms]
  currSyms,::syms;
  `tradeAgg upsert @[count[syms]#enlist initWith;`sym`time,cols rc;:;(syms;currTime),(initWith cols rc)^value flip rc:rollCache ([] sym: syms)];
 };

upd:{[tblName;data] updMinute[data] each exec distinct time from data:() xkey preprocess data};
updMinute:{[data;tm]
  if[tm<>currTime; roll tm; currTime::tm];
  data:select from data where time=tm;
  if[count msyms:syms where not (syms:data`sym)in currSyms; addSyms msyms];
  updateAgg[`tradeAgg;offset+currSyms?syms;data];
 };

परीक्षण

आइए सेवा के प्रदर्शन की जाँच करें। ऐसा करने के लिए, आइए इसे एक अलग प्रक्रिया में चलाएं (कोड को service.q फ़ाइल में डालें) और init फ़ंक्शन को कॉल करें:

q service.q –p 5566

q)init[]

दूसरे कंसोल में, दूसरी Q प्रक्रिया प्रारंभ करें और पहले से कनेक्ट करें:

h:hopen `:host:5566
h:hopen 5566 // если оба на одном хосте

सबसे पहले, आइए प्रतीकों की एक सूची बनाएं - 10000 टुकड़े और एक यादृच्छिक तालिका बनाने के लिए एक फ़ंक्शन जोड़ें। दूसरे कंसोल में:

syms:`IBM`AAPL`GOOG,-9997?`8
rnd:{[n;t] ([] sym:n?syms; time:t+asc n#til 25; price:n?10f; size:n?10)}

तालिका में उन्हें ढूंढना आसान बनाने के लिए मैंने सूची में तीन वास्तविक प्रतीक जोड़े। rnd फ़ंक्शन n पंक्तियों के साथ एक यादृच्छिक तालिका बनाता है, जहां समय t से t+25 मिलीसेकंड तक भिन्न होता है।

अब आप सेवा में डेटा भेजने का प्रयास कर सकते हैं (पहले दस घंटे जोड़ें):

{h (`upd;`trade;rnd[10000;x])} each `time$00:00 + til 60*10

आप सेवा में जाँच सकते हैं कि तालिका अद्यतन कर दी गई है:

c 25 200
select from tradeAgg where sym=`AAPL
-20#select from tradeAgg where sym=`AAPL

रिजल्ट:

sym|time|high|low|firstPrice|lastPrice|firstSize|lastSize|numTrades|volume|pvolume|turnover|avgPrice|avgSize|vwap|cumVolume
--|--|--|--|--|--------------------------------
AAPL|09:27|9.258904|9.258904|9.258904|9.258904|8|8|1|8|9.258904|74.07123|9.258904|8|9.258904|2888
AAPL|09:28|9.068162|9.068162|9.068162|9.068162|7|7|1|7|9.068162|63.47713|9.068162|7|9.068162|2895
AAPL|09:31|4.680449|0.2011121|1.620827|0.2011121|1|5|4|14|9.569556|36.84342|2.392389|3.5|2.631673|2909
AAPL|09:33|2.812535|2.812535|2.812535|2.812535|6|6|1|6|2.812535|16.87521|2.812535|6|2.812535|2915
AAPL|09:34|5.099025|5.099025|5.099025|5.099025|4|4|1|4|5.099025|20.3961|5.099025|4|5.099025|2919

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

tm:10:00:00.000
stressTest:{[n] 1 string[tm]," "; times,::h ({st:.z.T; upd[`trade;x]; .z.T-st};rnd[n;tm]); tm+:25}
start:{[n] times::(); do[4800;stressTest[n]]; -1 " "; `min`avg`med`max!(min times;avg times;med times;max times)}

4800 दो मिनट है. आप पहले प्रत्येक 1000 मिलीसेकंड में 25 पंक्तियाँ चलाने का प्रयास कर सकते हैं:

start 1000

मेरे मामले में, परिणाम प्रति अद्यतन कुछ मिलीसेकंड के आसपास है। तो मैं तुरंत पंक्तियों की संख्या 10.000 तक बढ़ा दूँगा:

start 10000

रिजल्ट:

min| 00:00:00.004
avg| 9.191458
med| 9f
max| 00:00:00.030

फिर, कुछ खास नहीं, लेकिन यह 24 मिलियन लाइनें प्रति मिनट, 400 हजार प्रति सेकंड है। 25 मिलीसेकंड से अधिक के लिए, अपडेट केवल 5 बार धीमा हुआ, जाहिर तौर पर जब मिनट बदल गया। आइए इसे बढ़ाकर 100.000 करें:

start 100000

रिजल्ट:

min| 00:00:00.013
avg| 25.11083
med| 24f
max| 00:00:00.108
q)sum times
00:02:00.532

जैसा कि आप देख सकते हैं, सेवा मुश्किल से सामना कर सकती है, लेकिन फिर भी यह चालू रहने में सफल रहती है। डेटा की इतनी मात्रा (प्रति मिनट 240 मिलियन पंक्तियाँ) बहुत बड़ी है; ऐसे मामलों में, सेवा के कई क्लोन (या यहां तक ​​कि दर्जनों क्लोन) लॉन्च करना आम बात है, जिनमें से प्रत्येक केवल वर्णों का हिस्सा संसाधित करता है। फिर भी, एक व्याख्या की गई भाषा के लिए परिणाम प्रभावशाली है जो मुख्य रूप से डेटा भंडारण पर केंद्रित है।

यह प्रश्न उठ सकता है कि प्रत्येक अद्यतन के आकार के साथ समय गैर-रैखिक रूप से क्यों बढ़ता है। कारण यह है कि श्रिंक फ़ंक्शन वास्तव में एक सी फ़ंक्शन है, जो अपडेटएजीजी की तुलना में बहुत अधिक कुशल है। एक निश्चित अद्यतन आकार (लगभग 10.000) से शुरू होकर, updateAgg अपनी सीमा तक पहुँच जाता है और फिर इसका निष्पादन समय अद्यतन आकार पर निर्भर नहीं करता है। यह प्रारंभिक चरण Q के कारण है कि सेवा इतनी मात्रा में डेटा को पचाने में सक्षम है। यह इस बात पर प्रकाश डालता है कि बड़े डेटा के साथ काम करते समय सही एल्गोरिदम चुनना कितना महत्वपूर्ण है। एक अन्य बिंदु मेमोरी में डेटा का सही भंडारण है। यदि डेटा को स्तंभ के अनुसार संग्रहीत नहीं किया गया था या समय के अनुसार ऑर्डर नहीं किया गया था, तो हम टीएलबी कैश मिस जैसी चीज़ से परिचित होंगे - प्रोसेसर एड्रेस कैश में मेमोरी पेज एड्रेस की अनुपस्थिति। असफल होने पर किसी पते की खोज करने में लगभग 30 गुना अधिक समय लगता है, और यदि डेटा बिखरा हुआ है, तो यह सेवा को कई गुना धीमा कर सकता है।

निष्कर्ष

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

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

स्रोत: www.habr.com