वास्तविक-समय सेवाको उदाहरण प्रयोग गरेर Q र KDB+ भाषाका सुविधाहरू

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

परिचय

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

यस लेखमा हामीले Q मा पूर्ण-विस्तारित कार्यक्रम लागू गर्छौं र तपाईंले यसलाई प्रयास गर्न सक्नुहुन्छ। यो गर्नको लागि, तपाईंलाई वास्तविक Q चाहिन्छ। तपाईंले kx कम्पनीको वेबसाइटमा नि:शुल्क ३२-बिट संस्करण डाउनलोड गर्न सक्नुहुन्छ - www.kx.com। त्यहाँ, यदि तपाइँ इच्छुक हुनुहुन्छ भने, तपाइँ Q मा सन्दर्भ जानकारी पाउनुहुनेछ, पुस्तक Q Mortals को लागी र यस विषयमा विभिन्न लेखहरू।

समस्याको गठन

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

सेवाले sym स्तम्भबाट प्रत्येक आगमन प्रतीकको लागि प्रत्येक मिनेट गणना गर्नु पर्छ समग्र प्रकार्यहरूको सेट - अधिकतम मूल्य, औसत मूल्य, योग आकार, आदि। उपयोगी जानकारी। सरलताको लागि, हामी मान्नेछौं कि सबै प्रकार्यहरू क्रमशः गणना गर्न सकिन्छ, अर्थात्। नयाँ मान प्राप्त गर्न, यो दुई नम्बरहरू जान्न पर्याप्त छ - पुरानो र आगमन मानहरू। उदाहरण को लागी, प्रकार्य अधिकतम, औसत, योग मा यो गुण छ, तर मध्य प्रकार्य छैन।

हामी आगमन डेटा स्ट्रिम समय अर्डर गरिएको छ भनेर पनि मान्नेछौं। यसले हामीलाई अन्तिम मिनेटमा मात्र काम गर्ने अवसर दिनेछ। अभ्यासमा, केहि अद्यावधिकहरू ढिलो भएमा हालको र अघिल्लो मिनेटहरूसँग काम गर्न सक्षम हुन पर्याप्त छ। सादगीको लागि, हामी यस मामलालाई विचार गर्दैनौं।

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

आवश्यक एकत्रीकरण कार्यहरू तल सूचीबद्ध छन्। मैले सेवामा लोड बढाउन सकेसम्म धेरैलाई लिएँ:

• उच्च - अधिकतम मूल्य - प्रति मिनेट अधिकतम मूल्य।
• कम - न्यूनतम मूल्य - न्यूनतम मूल्य प्रति मिनेट।
• पहिलो मूल्य - पहिलो मूल्य - प्रति मिनेट पहिलो मूल्य।
• अन्तिम मूल्य - अन्तिम मूल्य - प्रति मिनेट अन्तिम मूल्य।
• firstSize - पहिलो आकार - प्रति मिनेट पहिलो व्यापार आकार।
• अन्तिम आकार - अन्तिम आकार - एक मिनेटमा अन्तिम व्यापार आकार।
• numTrades - गणना i - प्रति मिनेट ट्रेडहरूको संख्या।
• मात्रा - योग आकार - प्रति मिनेट व्यापार आकार को योग।
• pvolume - योग मूल्य - प्रति मिनेट मूल्यहरूको योग, औसत मूल्यको लागि आवश्यक छ।
• - कुल कारोबार मूल्य * आकार - प्रति मिनेट लेनदेनको कुल मात्रा।
• avgPrice – pvolume%numTrades – औसत मूल्य प्रति मिनेट।
• avgSize – मात्रा%numTrades – औसत व्यापार आकार प्रति मिनेट।
• vwap - कारोबार% मात्रा - लेनदेन आकार द्वारा भारित प्रति मिनेट औसत मूल्य।
• cumVolume - योग मात्रा - सम्पूर्ण समय मा लेनदेन को संचित आकार।

तुरुन्तै एउटा गैर-स्पष्ट बिन्दुमा छलफल गरौं - कसरी यी स्तम्भहरू पहिलो पटक र प्रत्येक पछिको मिनेटको लागि सुरु गर्ने। पहिलो मूल्य प्रकारका केही स्तम्भहरू प्रत्येक पटक शून्य गर्न सुरु गर्नुपर्छ; अन्य भोल्युम प्रकारहरू सधैं 0 मा सेट हुनुपर्छ। त्यहाँ पनि स्तम्भहरू छन् जसलाई एक संयुक्त दृष्टिकोण चाहिन्छ - उदाहरणका लागि, cumVolume अघिल्लो मिनेटबाट प्रतिलिपि हुनुपर्छ, र पहिलो 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 объяснен ниже

मैले सुविधाको लागि शब्दकोशमा sym र समय थपेको छु, अब initWith अन्तिम समेकित तालिकाबाट तयार-निर्मित लाइन हो, जहाँ यो सही sym र समय सेट गर्न बाँकी छ। तपाइँ यसलाई तालिकामा नयाँ पङ्क्तिहरू थप्न प्रयोग गर्न सक्नुहुन्छ।

एग्रीगेशन प्रकार्य सिर्जना गर्दा हामीलाई aggCols चाहिन्छ। Q मा अभिव्यक्तिहरूको मूल्याङ्कन गरिएको क्रम (दायाँबाट बाँया) को कारणले सूची उल्टो हुनुपर्छ। लक्ष्य भनेको उच्च देखि कम भोल्युम सम्मको दिशामा गणना सुनिश्चित गर्नु हो, किनकि केही स्तम्भहरू अघिल्लोमा निर्भर हुन्छन्।

स्तम्भहरू जुन अघिल्लो एकबाट नयाँ मिनेटमा प्रतिलिपि गर्न आवश्यक छ, सुविधाको लागि सिम स्तम्भ थपिएको छ:

rollColumns:`sym`cumVolume;

अब स्तम्भहरूलाई कसरी अद्यावधिक गर्ने भन्ने अनुसार समूहहरूमा विभाजन गरौं। तीन प्रकार छुट्याउन सकिन्छ:

1. Accumulators (भोल्युम, टर्नओभर, ..) - हामीले अघिल्लो एकमा आगमन मान थप्नु पर्छ।
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

यो विधिको एक हानि छ - गणना गरिएको स्तम्भहरूको सेट पूर्वनिर्धारित छ। सौभाग्य देखि, Q मा, चयन पनि कार्यको रूपमा लागू गरिएको छ जहाँ तपाइँ गतिशील रूपमा सिर्जना गरिएका तर्कहरू प्रतिस्थापन गर्न सक्नुहुन्छ:

?[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 मा, लुप्सको सट्टा नक्सा/घटाउने प्रकार्यहरू प्रयोग गर्न सामान्य छ। तर 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 मा, सूची सिर्जना गर्ने एक प्रकार्य पनि हो र हामी यसलाई सूचीहरूको सूची प्राप्त गर्न प्रत्येक (नक्सा) प्रकार्य प्रयोग गरेर कल गर्न सक्छौं।

गणना गरिएको स्तम्भहरूको सेट निश्चित छैन भनेर सुनिश्चित गर्न, हामी माथिको अभिव्यक्ति गतिशील रूपमा सिर्जना गर्नेछौं। पङ्क्ति र inp चरहरू प्रयोग गरी एकत्रित र इनपुट डेटालाई सन्दर्भ गर्नका लागि प्रत्येक स्तम्भको गणना गर्न पहिले कार्यहरू परिभाषित गरौं:

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] स्तम्भ द्वारा यो पहिलो हो भनेर निर्धारण गर्न सक्छौं - यदि यसमा ० छ भने, मान पहिलो हो। Q सँग चयन प्रकार्य छ - ?[बुलियन सूची;सूची १;सूची २] - जसले पहिलो तर्कको अवस्थाको आधारमा सूची १ वा २ बाट मान चयन गर्छ:

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

यहाँ मैले मेरो प्रकार्यको साथ एक सामान्यीकृत असाइनमेन्ट (घुमिएको कोष्ठकहरूमा अभिव्यक्ति) बोलाएँ। यसले हालको मान (पहिलो तर्क) र एक अतिरिक्त तर्क प्राप्त गर्दछ, जुन मैले 4 प्यारामिटरमा पास गर्छु।

ब्याट्री स्पिकरहरू छुट्टै थपौं, किनभने तिनीहरूका लागि प्रकार्य समान छ:

// 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 प्रकार्य (Q सेवाहरूको लागि यस प्रकार्यको लागि परम्परागत नाम), जसलाई ग्राहकले डाटा थप्न बोलाइन्छ:

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];
 };

परीक्षण गर्दै

सेवाको कार्यसम्पादन जाँच गरौं। यो गर्नको लागि, यसलाई छुट्टै प्रक्रियामा चलाउनुहोस् (सेवा. 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 मिलियन पङ्क्तिहरू) अत्यन्त ठूलो छ, यस्तो अवस्थामा, सेवाको धेरै क्लोनहरू (वा दर्जनौं क्लोनहरू) सुरु गर्न सामान्य छ, जसमध्ये प्रत्येकले क्यारेक्टरहरूको मात्र भागलाई प्रशोधन गर्दछ। अझै, परिणाम एक व्याख्या गरिएको भाषाको लागि प्रभावशाली छ जुन मुख्य रूपमा डाटा भण्डारणमा केन्द्रित छ।

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

निष्कर्षमा

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

म नोट गर्नेछु कि माथिको Q ले के गर्न सक्छ भन्ने मात्र अंश हो, यसमा अन्य अद्वितीय सुविधाहरू पनि छन्। उदाहरणका लागि, एक अत्यन्त सरल IPC प्रोटोकल जसले व्यक्तिगत Q प्रक्रियाहरू बीचको सीमा मेटाउँछ र तपाईंलाई यी सयौं प्रक्रियाहरूलाई एकल नेटवर्कमा संयोजन गर्न अनुमति दिन्छ, जुन विश्वका विभिन्न भागहरूमा दर्जनौं सर्भरहरूमा अवस्थित हुन सक्छ।

स्रोत: www.habr.com