🥇रिअल-टाइम सेवेचे उदाहरण वापरून Q आणि KDB+ भाषेच्या शक्यता

KDB+ बेस, क्यू प्रोग्रामिंग भाषा काय आहेत, त्यांची ताकद आणि कमकुवतता काय आहेत याबद्दल तुम्ही माझ्या मागील लेखात वाचू शकता लेख आणि थोडक्यात परिचयात. लेखात, आम्ही Q वर एक सेवा कार्यान्वित करू जी येणार्‍या डेटा प्रवाहावर प्रक्रिया करेल आणि "रिअल टाइम" मोडमध्ये दर मिनिटाला विविध एकत्रीकरण कार्यांची गणना करेल (म्हणजे, डेटाच्या पुढील भागापूर्वी सर्वकाही मोजण्यासाठी वेळ असेल). Q चे मुख्य वैशिष्ट्य हे आहे की ही एक वेक्टर भाषा आहे जी तुम्हाला एकल ऑब्जेक्ट्ससह नाही तर त्यांच्या अॅरे, अॅरे आणि इतर जटिल ऑब्जेक्ट्ससह ऑपरेट करू देते. Q आणि त्याचे नातेवाईक K, J, APL यासारख्या भाषा त्यांच्या संक्षिप्ततेसाठी प्रसिद्ध आहेत. बर्‍याचदा, Java सारख्या परिचित भाषेत कोडच्या अनेक स्क्रीन्स घेणारा प्रोग्राम त्यावर काही ओळींमध्ये लिहिला जाऊ शकतो. या लेखात मला हेच दाखवायचे आहे.

परिचय

KDB+ हा एक स्तंभीय डेटाबेस आहे जो खूप मोठ्या प्रमाणात डेटावर केंद्रित आहे, विशिष्ट पद्धतीने (प्रामुख्याने वेळेनुसार) ऑर्डर केला जातो. हे प्रामुख्याने वित्तीय संस्थांमध्ये वापरले जाते - बँका, गुंतवणूक निधी, विमा कंपन्या. Q भाषा ही KDB+ ची अंतर्गत भाषा आहे जी तुम्हाला या डेटासह प्रभावीपणे कार्य करण्यास अनुमती देते. Q विचारधारा संक्षिप्तता आणि कार्यक्षमता आहे, तर स्पष्टतेचा त्याग केला जातो. हे या वस्तुस्थितीद्वारे न्याय्य आहे की वेक्टर भाषा कोणत्याही परिस्थितीत समजणे कठीण होईल आणि रेकॉर्डिंगची संक्षिप्तता आणि समृद्धता आपल्याला एका स्क्रीनवर प्रोग्रामचा बराच मोठा भाग पाहण्याची परवानगी देते, जे शेवटी समजून घेणे सोपे करते.

या लेखात आम्ही Q मध्ये एक पूर्ण-प्रक्रिया कार्यक्रम लागू करतो आणि तुम्हाला कदाचित तो वापरून पहावासा वाटेल. हे करण्यासाठी, तुम्हाला वास्तविक Q आवश्यक असेल. तुम्ही kx कंपनीच्या वेबसाइटवर मोफत 32-बिट आवृत्ती डाउनलोड करू शकता – www.kx.com. तेथे, तुम्हाला स्वारस्य असल्यास, तुम्हाला Q, पुस्तकावरील संदर्भ माहिती मिळेल मर्त्यांसाठी Q आणि या विषयावरील विविध लेख.

समस्येची निर्मिती

एक स्रोत आहे जो प्रत्येक 25 मिलीसेकंदांनी डेटासह टेबल पाठवतो. KDB+ चा वापर प्रामुख्याने फायनान्समध्ये होत असल्याने, आम्ही असे गृहीत धरू की हे व्यवहारांचे (ट्रेड) सारणी आहे, ज्यामध्ये खालील स्तंभ आहेत: वेळ (वेळ मिलिसेकंदमध्ये), sym (स्टॉक एक्स्चेंजवरील कंपनी पदनाम - IBM, AAPL,…), किंमत (ज्या किंमतीला शेअर्स खरेदी केले होते), आकार (व्यवहाराचा आकार). 25 मिलीसेकंद मध्यांतर अनियंत्रित आहे, खूप लहान नाही आणि खूप लांब नाही. त्याच्या उपस्थितीचा अर्थ असा आहे की डेटा आधीच बफर केलेल्या सेवेवर येतो. सध्याच्या लोडवर अवलंबून डायनॅमिक बफरिंगसह, सेवेच्या बाजूने बफरिंग लागू करणे सोपे होईल, परंतु साधेपणासाठी, आम्ही एका निश्चित अंतरावर लक्ष केंद्रित करू.

सेवेने सिम स्तंभातील प्रत्येक इनकमिंग चिन्हासाठी प्रत्येक मिनिटाला एकत्रित फंक्शन्सचा संच मोजला पाहिजे - कमाल किंमत, सरासरी किंमत, बेरीज आकार इ. उपयुक्त माहिती. साधेपणासाठी, आम्ही असे गृहीत धरू की सर्व फंक्शन्सची वाढीव गणना केली जाऊ शकते, म्हणजे. नवीन मूल्य प्राप्त करण्यासाठी, दोन संख्या जाणून घेणे पुरेसे आहे - जुनी आणि येणारी मूल्ये. उदाहरणार्थ, फंक्शन्स कमाल, सरासरी, बेरीजमध्ये हा गुणधर्म असतो, परंतु मध्य फंक्शनमध्ये नाही.

आम्ही येणारा डेटा प्रवाह वेळ क्रमाने आहे असे गृहीत धरू. यामुळे केवळ शेवटच्या क्षणी काम करण्याची संधी मिळेल. सराव मध्ये, काही अद्यतने उशीर झाल्यास वर्तमान आणि मागील मिनिटांसह कार्य करण्यास सक्षम असणे पुरेसे आहे. साधेपणासाठी, आम्ही या प्रकरणात विचार करणार नाही.

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

आवश्यक एकत्रीकरण कार्ये खाली सूचीबद्ध आहेत. सेवेवरील भार वाढवण्यासाठी मी त्यापैकी शक्य तितके घेतले:

उच्च - कमाल किंमत - प्रति मिनिट कमाल किंमत.
कमी - किमान किंमत - प्रति मिनिट किमान किंमत.
पहिली किंमत – पहिली किंमत – पहिली किंमत प्रति मिनिट.
शेवटची किंमत - शेवटची किंमत - प्रति मिनिट शेवटची किंमत.
प्रथम आकार – प्रथम आकार – प्रति मिनिट प्रथम व्यापार आकार.
lastSize - शेवटचा आकार - एका मिनिटात शेवटचा व्यापार आकार.
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;

आता स्तंभांना ते कसे अपडेट करायचे त्यानुसार गटांमध्ये विभागू. तीन प्रकार ओळखले जाऊ शकतात:

संचयक (व्हॉल्यूम, टर्नओव्हर,..) - आपण येणारे मूल्य मागील एकामध्ये जोडले पाहिजे.
एका विशेष बिंदूसह (उच्च, निम्न, ..) - मिनिटातील पहिले मूल्य येणार्‍या डेटामधून घेतले जाते, उर्वरित फंक्शन वापरून मोजले जातात.
उर्वरित. नेहमी फंक्शन वापरून गणना केली जाते.

चला या वर्गांसाठी व्हेरिएबल्स परिभाषित करूया:

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

स्पष्टतेसाठी, मी पार्स फंक्शन वापरले, जे क्यू एक्स्प्रेशनसह स्ट्रिंगला व्हॅल्यूमध्ये बदलते जे 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 मध्ये, loops ऐवजी map/reduce फंक्शन्स वापरणे सामान्य आहे. परंतु क्यू ही सदिश भाषा असल्याने आणि आपण सर्व चिन्हांवर एकाच वेळी सर्व क्रिया सहजपणे लागू करू शकतो, नंतर प्रथम अंदाजेपर्यंत आपण लूपशिवाय अजिबात करू शकतो, सर्व चिन्हांवर एकाच वेळी ऑपरेशन्स करू शकतो:

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] स्तंभानुसार ते पहिले आहे हे निर्धारित करू शकतो - जर त्यात 0 असेल, तर मूल्य पहिले आहे. Q मध्ये सिलेक्ट फंक्शन आहे - ?[बूलियन लिस्ट;लिस्ट1;लिस्ट2] - जे पहिल्या वितर्कातील कंडिशननुसार सूची 1 किंवा 2 मधून मूल्य निवडते:

// 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 फंक्शन (क्यू सेवांसाठी या फंक्शनचे पारंपारिक नाव), ज्याला क्लायंट डेटा जोडण्यासाठी कॉल करतो:

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 म्हणजे दोन मिनिटे. तुम्ही प्रत्येक २५ मिलीसेकंदांनी १००० पंक्तींसाठी प्रथम धावण्याचा प्रयत्न करू शकता:

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 दशलक्ष ओळी प्रति मिनिट) अत्यंत मोठी आहे; अशा प्रकरणांमध्ये, सेवेचे अनेक क्लोन (किंवा अगदी डझनभर क्लोन) लाँच करणे सामान्य आहे, ज्यापैकी प्रत्येक वर्णांच्या केवळ भागावर प्रक्रिया करतो. तरीही, मुख्यतः डेटा स्टोरेजवर लक्ष केंद्रित करणार्‍या व्याख्या केलेल्या भाषेसाठी परिणाम प्रभावी आहे.

प्रश्न उद्भवू शकतो की प्रत्येक अद्यतनाच्या आकारानुसार वेळ नॉन-रेखीय का वाढतो. याचे कारण म्हणजे shrink फंक्शन हे C फंक्शन आहे, जे updateAgg पेक्षा जास्त कार्यक्षम आहे. एका विशिष्ट अद्यतन आकारापासून (सुमारे 10.000) प्रारंभ करून, updateAgg त्याच्या कमाल मर्यादेपर्यंत पोहोचते आणि नंतर त्याची अंमलबजावणी वेळ अद्यतन आकारावर अवलंबून नसते. प्राथमिक पायरी Q मुळे ही सेवा इतक्या प्रमाणात डेटा पचवण्यास सक्षम आहे. मोठ्या डेटासह काम करताना योग्य अल्गोरिदम निवडणे किती महत्त्वाचे आहे हे हे हायलाइट करते. दुसरा मुद्दा म्हणजे मेमरीमधील डेटाचे योग्य संचयन. जर डेटा स्तंभीयरित्या संग्रहित केला गेला नसेल किंवा वेळेनुसार ऑर्डर केला गेला नसेल, तर आम्ही TLB कॅशे मिस - प्रोसेसर अॅड्रेस कॅशेमध्ये मेमरी पृष्ठ पत्ता नसणे यासारख्या गोष्टींशी परिचित होऊ. पत्ता शोधण्यात अयशस्वी झाल्यास सुमारे 30 पट जास्त वेळ लागतो आणि डेटा विखुरलेला असल्यास, यामुळे सेवा अनेक वेळा कमी होऊ शकते.

निष्कर्ष

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

मी लक्षात घेईन की वरील हा फक्त Q काय करू शकतो त्याचा एक भाग आहे, त्यात इतर अद्वितीय वैशिष्ट्ये देखील आहेत. उदाहरणार्थ, एक अत्यंत साधा IPC प्रोटोकॉल जो वैयक्तिक Q प्रक्रियांमधील सीमा पुसून टाकतो आणि तुम्हाला यापैकी शेकडो प्रक्रिया एकाच नेटवर्कमध्ये एकत्र करू देतो, जे जगाच्या विविध भागांमध्ये डझनभर सर्व्हरवर असू शकतात.

स्त्रोत: www.habr.com

रिअल-टाइम सेवेचे उदाहरण वापरून Q आणि KDB+ भाषेची वैशिष्ट्ये

परिचय

समस्येची निर्मिती

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

गणना क्रम

अंतिम टप्पे

चाचणी

निष्कर्ष

एक टिप्पणी जोडा Отменить ответ