Arról, hogy mi a KDB+ adatbázis, a Q programozási nyelv, és mik a fürtéi és gyengeségei, előző írásomban olvashat. Röviden az előszóban. A cikkben megvalósítunk egy Q alapú szolgáltatást, amely a bejövő adatfolyamot kezeli és valós időben számít különböző aggregáló funkciókat percenként (azaz meg tudja számítani mindent a következő adatcsomag előtt). A Q fő jellemzője, hogy vektoralapú nyelv, amely lehetővé teszi az egyedi objektumok helyett azok tömbjeivel, tömbök tömbjeivel és egyéb összetett objektumokkal való munkát. Az olyan nyelvek, mint a Q, valamint a hozzátartozó K, J, APL híresek a tömörségükről. Gyakran előfordul, hogy egy Java típusú, több képernyyt igénylő programot pár sorban meg lehet írni rajtuk. Ezt szeretném bemutatni ebben a cikkben.
Bevezetés
A KDB+ egy oszlop alapú adatbázis, amelyet nagyon nagy adatvolumenek kezelésére terveztek, amelyeket adott módon (elsősorban idő szerint) rendeznek. Elsősorban pénzügyi intézményekben - bankokban, befektetési alapokban, biztosítótársaságokban használják. A Q nyelv a KDB+ belső nyelve, amely lehetővé teszi az adatokkal való hatékony munkát. A Q ideológiája a tömörség és hatékonyság, miközben a világosság érdekében áldozatokat hoznak. Ezt azzal indokolják, hogy a vektoralapú nyelv bármi esetre nehezen érthető, míg a tömörség és sűrűség a program sokkal nagyobb részét mutatja egy képernyőn, ez végső soron megkönnyíti annak megértését.
A cikkben egy teljes Q programot valósítunk meg, és lehet, hogy kedve támad kipróbálni. Ehhez szüksége lesz magára a Q-ra. A cég kx honlapjáról töltheti le a 32 bites ingyenes verziót - . Ott ugyanitt, ha érdekli, a Q-val kapcsolatos referencia anyagokat, a "Q For Mortals" című könyvet Van egy forrás, amely 25 milliszekundumonként egy táblázatot küld az adatokkal. Mivel a KDB+ elsősorban a pénzügyekben használatos, tegyük fel, hogy ez egy kereskedési (trades) táblázat, amely a következő oszlopokat tartalmazza: time (idő milliszekundumban), sym (a cég tőzsdei jelölése -
Lépjünk át a kitűzött feladatra: írjunk egy XDP mechanizmust SYN cookie-kkal.
AAPL IBM, AAPL,…), ár (ár, amelyen a részvényeket megvásárolták), méret (ügylet mérete). A 25 milliszekundumos időköz véletlenszerűen lett kiválasztva, nem túl kicsi és nem túl nagy. Azt jelzi, hogy az adatok már puffertárolva érkeznek a szolgáltatásba. Könnyen megvalósítható lenne a puffertárolás a szolgáltatás oldalán, beleértve a dinamikusat is, amely a pillanatnyi terheléstől függ, de az egyszerűség kedvéért ragaszkodjunk a fix időközhez.
A szolgáltatásnak percenként kell számítania minden érkező karaktert a sym oszlopból, aggregáló függvények halmazát – max ár, átlag ár, összeg méret stb. hasznos információt. Az egyszerűség kedvéért feltételezzük, hogy minden függvényt inkrementálisan lehet kiszámítani, azaz az új értékhez elegendő két számot ismerni – a régit és az új érkező értéket. Például a max, az átlag, az összeg függvények rendelkeznek ezzel a tulajdonsággal, de a medián függvény nem.
Továbbá feltételezzük, hogy az érkező adatfolyam időrendben van. Ez lehetővé teszi számunkra, hogy csak az utolsó perccel dolgozzunk. A gyakorlatban elegendő kezelni az aktuális és az előző percet arra az esetre, ha néhány frissítés késlekedne. Az egyszerűség kedvéért nem vizsgáljuk ezt az esetet.
Aggregáló függvények
Az alábbiakban felsoroljuk a szükséges aggregáló függvényeket. Minél többet gyűjtöttem össze, hogy növeljem a szolgáltatás terhelését:
- high – max ár – maximális ár percenként.
- low – min ár – minimális ár percenként.
- firstPrice – első ár – első ár percenként.
- lastPrice – utolsó ár – utolsó ár percenként.
- firstSize – első méret – első ügylet méret percenként.
- lastSize – utolsó méret – utolsó ügylet méret percenként.
- numTrades – szám i – ügyletek száma percenként.
- volume – összeg méret – ügyletek méretének összege percenként.
- pvolume – összeg ár – árak összege percenként, szükséges az avgPrice-hoz.
- turnover – összeg ár*méret – összes ügylet volumene percenként.
- avgPrice – pvolume%numTrades – átlag ár percenként.
- avgSize – volume%numTrades – átlagos ügylet mérete percenként.
- vwap – turnover%volume – a méretre súlyozott átlag ár percenként.
- cumVolume – összeg volume – az összes időszakra vonatkozó ügylet mérete.
Beszéljünk egy nem egyértelmű pontról – hogyan inicializáljuk ezeket az oszlopokat először és minden egyes következő percben. Egyes oszlopokat, mint a firstPrice, minden alkalommal null értékkel kell inicializálni, mert az értékük nem meghatározott. Mások, mint a volume, mindig 0-ra kell beállítani. Vannak olyan oszlopok is, amelyek kombinált megközelítést igényelnek – például a cumVolume értékét át kell másolni az előző percből, míg az első esetben 0-ra kell állítani. Mindezeket a paramétereket szótár adat típussal fogjuk megadni (analóg a rekorddal):
// 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 объяснен ниже
Hozzáadtam a sym és time elemeket a szótárhoz a kényelem kedvéért, így az initWith most már a végső aggregált táblázatból származó kész sor, ahol csak a megfelelő sym és time értéket kell megadni. Ezt fel lehet használni új sorok hozzáadásához a táblázathoz.
Az aggCols-ra szükségünk lesz az aggregáló függvény létrehozásakor. A listát meg kell fordítani a Q kifejezések számítási sorrendjének sajátosságai miatt (jobb oldalról balra). A cél az, hogy a számítást a high-tól a cumVolume felé irányítsuk, mivel egyes oszlopok az előzőktől függenek.
Az oszlopok, amelyeket át kell másolni az új percre, a sym oszlop a kényelem kedvéért lett hozzáadva:
rollColumns:`sym`cumVolume;
Most csoportosítjuk az oszlopokat aszerint, hogyan kell őket frissíteni. Három típust lehet megkülönböztetni:
- Akumulátorok (volume, turnover,..) – hozzá kell adni a bejövő értéket az előzőhöz.
- Különleges pontok (high, low, ..) – az első érték a percben a bejövő adatokból származik, a többit funkciók segítségével számítjuk ki.
- A többiek. Mindig funkciók segítségével számítjuk.
Határozzuk meg a változókat ezekhez a csoportokhoz:
accumulatorCols:`numTrades`volume`pvolume`turnover;
specialCols:`high`low`firstPrice`firstSize;
A számítások sorrendje
Az aggregált táblázatot két lépésben frissítjük. A hatékonyság érdekében először tömörítjük a bejövő táblázatot, hogy minden szimbólumhoz és perchez egy sor maradjon. Az, hogy minden funkciónk inkrementális és asszociatív, garantálja, hogy ennek az extra lépésnek az eredménye nem változik. A táblázat tömörítése kiválható lenne egy select segítségével:
select high:max price, low:min price … by sym,time.minute from table
Ennek a módszernek van egy hátránya – a számított oszlopok halmaza előre meg van határozva. Szerencsére a Q-ban a select funkcióként is megvalósított, ami lehetővé teszi dinamikusan létrehozott argumentumok bevitelét:
?[table;whereClause;byClause;selectClause]
Nem fogom részletesen leírni az argumentumok formátumát, a mi esetünkben nem trivialisak csak a by és select kifejezések, és ezek szótárak kell legyenek columns!expressions formátumban. Így a tömörítő függvényt így lehet megadni:
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 ez a map függvény a Q nyelvben egy lista számára
preprocess:?[;();`sym`time!`sym`time.minute;selExpression];
A világosság kedvéért használtam a parse függvényt, amely a Q kifejezéssel rendelkező sztringet egy olyan értékké alakítja, amelyet át lehet adni az eval függvénynek és amely szükséges a funkcionális selecthez. Érdemes megjegyezni, hogy a preprocess a select függvény projekciójaként van megadva (azaz részben meghatározott argumentumokkal rendelkező függvény), hiányzik egy argumentum (táblázat). Ha a preprocess-t alkalmazzuk a táblázatra, akkor egy tömörített táblázatot kapunk.
A második lépés az aggregált táblázat frissítése. Először írjuk le az algoritmust pseudokódban:
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];
…
A Q nyelvben a ciklusok helyett általában map/reduce függvényeket használnak. De mivel a Q vektor nyelv, és minden műveletet nyugodtan alkalmazhatunk az összes szimbolikusan egyszerre, kezdetben elkerülhetjük a ciklust, és közvetlenül az összes szimbolikusan hajthatjuk végre a műveleteket:
idx:calcIdx inputTable;
row:aggTable idx;
aggTable[idx;`high]: row[`high] | inputTable`high;
aggTable[idx;`volume]: row[`volume] + inputTable`volume;
…
De tovább is léphetünk, a Q nyelvnek van egy egyedülálló és rendkívül hatékony operátora – a generalizált hozzárendelés operátora. Ez lehetővé teszi az értékek halmazának megváltoztatását a bonyolult adatszerkezetben indexek, függvények és argumentumok listájának felhasználásával. Ebben az esetben így néz ki:
idx:calcIdx inputTable;
rows:aggTable idx;
// .[target;(idx0;idx1;..);function;argument] ~ target[idx 0;idx 1;…]: function[target[idx 0;idx 1;…];argument], a mi esetünkben a függvény – a hozzárendelés
.[aggTable;(idx;aggCols);:;flip (row[`high] | inputTable`high;row[`volume] + inputTable`volume;…)];
Sajnos a táblázatban történő hozzárendeléshez sorok listájára van szükség, nem oszlopokra, és a mátrix transzponálásához (oszlopok listáját sorok listájává alakítva) flip függvényt kell alkalmazni. Nagy táblázatok esetén ez nem gazdaságos, ezért inkább az egyes oszlopokat külön-külön alkalmazzuk a generalizált hozzárendeléshez map függvény használatával (amely úgy néz ki, mint egy aposztróf):
[aggTable;;:;]'[(idx;)each aggCols; (row[`high] | inputTable`high;row[`volume] + inputTable`volume;…)];
Újra használjuk a függvény projekcióját. Érdemes megjegyezni, hogy a Q-ban a lista létrehozása szintén egy függvény, amelyet a each(map) függvény segítségével hívhatunk meg, hogy listát kapjunk listákból.
Annak érdekében, hogy a számított oszlopok készlete ne legyen rögzített, dinamikusan létrehozzuk a kifejezést. Először definiáljuk a számító függvényeket minden oszlophoz, a row és inp változókat felhasználva az aggregált és bemeneti adatok hivatkozására:
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");
Néhány oszlop különleges, az első értéküket nem szabad a függvény számolnia. Meghatározhatjuk, hogy az első az row[`numTrades] oszlopban van – ha 0 van benne, akkor az érték az első. A Q-nak van egy választó függvénye — ?[Boolean list;list1;list2] – amely az első argumentumban szereplő feltétel szerint választ egy értéket az 1-es vagy 2-es listából:
// high -> ?[isFirst;inp`high;row[`high]|inp`high]
// @ - тоже обобщенное присваивание для случая когда индекс неглубокий
@[`aggExpression;specialCols;{[x;y]"?[isFirst;inp`",y,";",x,"]"};string specialCols];
Itt általános hozzárendelést használtam a függvényemmel (a kifejezés kapcsos zárójelek között). Az aktuális érték (első argumentum) és egy további argumentum kerül átadásra, amelyet a 4. paraméterben adok meg.
Külön hozzáadunk akkumulator oszlopokat, mivel számukra a függvény ugyanaz:
// volume -> row[`volume]+inp`volume
aggExpression[accumulatorCols]:{"row[`",x,"]+inp`",x } each string accumulatorCols;
Ez egy tipikus hozzárendelés Q-ban, csak az értéklista azonnali hozzárendelésével. Végül létrehozzuk a fő függvényt:
// ":",/: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),")]}";
Ezzel a kifejezéssel dinamikusan hozok létre egy függvényt egy sorozatból, amely tartalmazza a fenti kifejezést. Az eredmény ennek így fog kinézni:
{[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])]}
Az oszlopok számításának sorrendje megfordított, mivel a Q-ban a sorrend jobbról balra történik.
Most két fő függvényünk van a számításokhoz, és már csak néhány infrastruktúrát kell hozzáadni, hogy a szolgáltatás készen álljon.
Végső lépések
Megvannak a preprocess és updateAgg függvényeink, amelyek elvégzik az összes munkát. De biztosítani kell a megfelelő átmenetet perceken keresztül, és ki kell számolni az aggregációs indexeket. Első lépésként határozzuk meg az init függvényt:
init:{
tradeAgg:: 0#enlist[initWith]; // létrehozunk egy üres típusú táblát, az enlist szótárat táblázattá alakít, a 0# pedig azt jelenti, hogy 0 elemet veszünk belőle
currTime::00:00; // kezdjük a 0-val, :: azt jelenti, hogy hozzárendelés egy globális változóhoz
currSyms::`u#`symbol$(); // `u# - listát fát alakít, a elemek keresésének felgyorsítása érdekében
offset::0; // index a tradeAgg-ben, ahol az aktuális perc kezdődik
rollCache:: `sym xkey update `u#sym from rollColumns#tradeAgg; // cache az utolsó roll oszlopok értékeiről, tábla sym kulccsal
}
Akkor definiáljuk a roll függvényt, amely megváltoztatja az aktuális percet:
roll:{[tm]
if[currTime>tm; :init[]]; // ha átléptük éjfélt, egyszerűen meghívjuk az init-et
rollCache,::offset _ rollColumns#tradeAgg; // frissítjük a cache-t – vegyük a roll oszlopokat az aggTable-ből, vágjuk le, és illesszük be a rollCache-be
offset::count tradeAgg;
currSyms::`u#`$();
}
Szükségünk lesz egy függvényre új szimbólumok hozzáadására:
addSyms:{[syms]
currSyms,::syms; // hozzáadjuk a jól ismert listahez
// hozzáadjuk a szimbólum, idő és rollColumns táblához az általános hozzárendelés felhasználásával.
// A ^ függvény alapértelmezett értékeket helyez el a roll oszlopok számára, ha a szimbólum nincs a cache-ben. A value flip táblázat visszaadja a táblázat oszlopainak listáját.
`tradeAgg upsert @[count[syms]#enlist initWith;`sym`time,cols rc;:;(syms;currTime), (initWith cols rc)^value flip rc:rollCache ([] sym: syms)];
}
És végül, az upd függvény (a Q szolgáltatásoknál hagyományos név), amelyet a kliens hív meg adatok hozzáadására:
upd:{[tblName;data] // tblName-re nincs szükségünk, de a szolgáltatás általában több táblát kezel
tm:exec distinct time from data:() xkey preprocess data; // előfeldolgozás & calc idő
updMinute[data] each tm; // hozzáadjuk az adatokat minden perc esetén
};
updMinute:{[data;tm]
if[tmcurrTime; roll tm; currTime::tm]; // megváltoztatjuk a percet, ha szükséges
data:select from data where time=tm; // szűrés
if[count msyms:syms where not (syms:data`sym)in currSyms; addSyms msyms]; // új szimbólumok
updateAgg[`tradeAgg;offset+currSyms?syms;data]; // frissítjük az aggregált táblát. A ? függvény az elemek indexét keresi a baloldali listában a jobboldali listában.
};
És ez mind. Íme a szolgáltatásunk teljes kódja, ahogy ígértük, mindössze néhány sor:
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];
};
Tesztelés
Ellenőrizzük a szolgáltatás teljesítményét. Ehhez indítsuk el egy külön folyamatban (helyezze a kódot a service.q fájlba) és hívja meg az init függvényt:
q service.q –p 5566
q)init[]
Egy másik konzolon indítson el egy második Q folyamatot, és csatlakozzon az elsőhöz:
h:hopen `:host:5566
h:hopen 5566 \/\/ ha mindkettő ugyanazon a gazdán van
Először hozzunk létre egy 10000 szimbólumból álló listát, és adjunk hozzá egy függvényt a véletlenszerű táblázat létrehozásához. A második konzolon:
syms:`IBM`AAPL`GOOG,-9997?`8
rnd:{[n;t] ([] sym:n?syms; time:t+asc n#til 25; price:n?10f; size:n?10)}
Hozzáadtam a szimbólumok listájához három valóságos elemet, hogy könnyebb legyen őket megtalálni a táblázatban. A rnd függvény véletlenszerű táblázatot készít n sorral, ahol az idő t-től t+25 milliszekundumig változik.
Most megpróbálhatjuk elküldeni az adatokat a szolgáltatásba (hozzáadjuk az első tíz órát):
{h (`upd;`trade;rnd[10000;x])} each `time$00:00 + til 60*10
Ellenőrizhetjük a szolgáltatásban, hogy a táblázat frissült:
c 25 200
select from tradeAgg where sym=`AAPL
-20#select from tradeAgg where sym=`AAPL
Eredmény:
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|2919Most kezdünk el terheléses tesztelést, hogy megtudjuk, mennyi adatot tud feldolgozni a szolgáltatás percenként. Emlékeztetnék, hogy 25 milliszekundumra állítottuk be az frissítési intervallumot. Ennek megfelelően a szolgáltatásnak (átlagosan) legalább 20 milliszekundum alatt kell végeznie egy frissítést, hogy időt adjon a felhasználóknak az adatok lekérésére. Írja be a következőt a második folyamatba:
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 – ez két perc. Próbáljuk meg először 1000 sort indítani 25 milliszekundumonként:
start 1000
Az én esetemben az eredmény körülbelül pár milliszekundum frissítésenként. Így azonnal növelni fogom a sorok számát 10.000-ra:
start 10000
Eredmény:
min| 00:00:00.004
avg| 9.191458
med| 9f
max| 00:00:00.030
Ismét nem mondható túl különösnek, pedig ez 24 millió sor percenként, 400 ezer másodpercenként. Több mint 25 milliszekundum frissítés lassult csak ötször, nyilván a percváltás miatt. Növeljük 100.000-re:
start 100000
Eredmény:
min| 00:00:00.013
avg| 25.11083
med| 24f
max| 00:00:00.108
q)sum times
00:02:00.532
Ahogyan látjuk, a szolgáltatás alig boldogul, de mindazonáltal sikerül a felszínen maradnia. Az ilyen adatmennyiség (240 millió sor percenként) rendkívül nagy, ilyen esetekben gyakran több klónt indítanak (vagy akár tucatnyi klónt) a szolgáltatásból, amelyek mindegyike csak egy részét dolgozza fel a karaktereknek. Ennek ellenére az eredmény lenyűgöző értelmezhető nyelv szempontjából, amely elsődlegesen az adatok tárolására orientált.
Felmerülhet a kérdés, hogy miért nő az idő nemlineárisan minden frissítés méretével együtt. Ennek oka, hogy a tömörítési függvény valójában egy C függvény, amely sokkal hatékonyabban működik, mint az updateAgg. Egy bizonyos frissítési méret (körülbelül 10.000) után az updateAgg eléri a plafonját, és a futási ideje már nem függ a frissítés méretétől. Pontosan a Q előlépésének köszönhetően a szolgáltatás képes feldolgozni az ilyen mennyiségű adatot. Ez hangsúlyozza, mennyire fontos, hogy nagy adatokkal dolgozva a megfelelő algoritmust válasszuk. Egy másik lényeges dolog az adatok helyes tárolása a memóriában. Ha az adatok nem oszloposan volnának tárolva, vagy nem lennének időben rendezve, akkor találkozhatnánk olyan dologgal, mint a TLB cache miss – a memóriacím lapjának hiánya a processzor címkészletében. Egy cím keresése körülbelül 30-szor tovább tart, ha nem sikerül, és a szétszórt adatok miatt a szolgáltatást többszörösen lelassíthatja.
Zárás
Ebben a cikkben azt mutattam be, hogy a KDB+ és a Q adatbázis nemcsak nagy adatok tárolására és egyszerű lekérdezések végrehajtására alkalmas, hanem adatfeldolgozó szolgáltatások létrehozására is, amelyek akár több százmillió sort / gigabájt adatot is képesek kezelni egyetlen Q folyamaton belül. Magának a Q nyelvnek köszönhetően rendkívül tömören és hatékonyan lehet adatfeldolgozási algoritmusokat megvalósítani a vektoriális jellegének, az SQL dialektus beépített értelmezőjének és a kiváló könyvtári funkciók készletének köszönhetően.
Megjegyzem, hogy a fentiek csak a Q lehetőségeinek egy részét képezik, vannak más egyedi jellemzői is. Például rendkívül egyszerű IPC protokollja, amely eltünteti a határt az egyes Q folyamatok között, és lehetővé teszi ezeknek a folyamatoknak a több százas csoportosítását egyetlen hálózatba, amely akár több tucat szerveren is elhelyezkedhet a világ különböző pontjain.
Forrás: habr.com
