Om hvad KDB+ er, programmeringssproget Q, og hvilke styrker og svagheder de har, kan læses i min tidligere artikel. og kort i introduktionen. I denne artikel vil vi implementere en tjeneste i Q, som vil behandle den indkommende datastream og beregne forskellige aggregerede funktioner i realtid (dvs. den vil nå at beregne alt, inden den næste datadump kommer). Hovedtræk ved Q er, at det er et vektorbaseret sprog, der gør det muligt at arbejde med ikke blot individuelle objekter, men også med arrays, arrays af arrays og andre komplekse objekter. Sprog som Q og beslægtede sprog K, J, APL er berømte for deres korthed. Ofte kan et program, der optager flere skærme med kode på et bekendt sprog som Java, skrives på dem i flere linjer. Det er netop det, jeg vil demonstrere i denne artikel.
Introduktion
KDB+ er en kolonnedatabase, der er designet til meget store datamængder, der er ordnet på en bestemt måde (primært efter tid). Den bruges primært i finansielle institutioner — banker, investeringsfonde, forsikringsselskaber. Q-sproget er det interne sprog for KDB+, der gør det muligt at arbejde effektivt med disse data. Ideologien bag Q er kortfattethed og effektivitet, mens forståeligheden ofres. Baggrunden for dette er, at et vektor-based sprog altid vil være svært at opfatte, og korthed samt tæthed i notationen giver mulighed for at se en meget større del af programmet på én skærm, hvilket i sidste ende letter forståelsen.
I artiklen vil vi implementere et fuldt program i Q, og du vil måske have lyst til at prøve det i praksis. Til det skal du bruge selve Q. Den gratis 32-bit version kan downloades fra kx's hjemmeside - . Der finder du også, hvis du er interesseret, referenceoplysninger om Q, bogen og forskellige artikler om emnet.
Problemformulering
Der er en kilde, der hvert 25. sekund sender en tabel med data. Da KDB+ primært anvendes inden for finanssektoren, lader os antage, at dette er en handels-tabel (trades), hvor der er følgende kolonner: time (tid i millisekunder), sym (forkortelse for selskabet på børsen - IBM, AAPL,…), price (prisen, hvormed aktierne er købt), size (størrelse på handlen). Intervallet på 25 millisekunder er valgt vilkårligt; det er hverken for småt eller for stort. Dets tilstedeværelse betyder, at dataene allerede er buffede i servicen. Det ville være nemt at implementere buffering på servicesiden, herunder dynamisk afhængig af den nuværende belastning, men for enkelhedens skyld stopper vi ved et fast interval.
Servicen skal for hver indkommende tegn fra kolonnen sym beregne et sæt aggregeringsfunktioner – max price, avg price, sum size osv. nyttig information. For enkelhedens skyld antager vi, at alle funktioner kan beregnes inkrementelt, dvs. for at opnå en ny værdi er det tilstrækkeligt at kende to tal – det gamle og det indkommende værdi. For eksempel har funktionerne max, average, sum denne egenskab, men medianfunktionen har ikke.
Vi antager også, at den indkommende datastream er ordnet efter tid. Dette giver os mulighed for kun at arbejde med det sidste minut. I praksis er det tilstrækkeligt at kunne arbejde med det nuværende og det foregående minut i tilfælde af, at nogle opdateringer er forsinkede. For enkelhedens skyld vil vi ikke overveje denne situation.
Aggregeringsfunktioner
Nedenfor er de nødvendige aggregeringsfunktioner listet. Jeg har taget så mange som muligt for at øge belastningen på servicen:
- high – max price – maksimal pris per minut.
- low – min price – minimal pris per minut.
- firstPrice – first price – første pris per minut.
- lastPrice – last price – sidste pris per minut.
- firstSize – first size – første størrelse på handlen per minut.
- lastSize – last size — sidste størrelse på handlen per minut.
- numTrades – count i – antal handler per minut.
- volume – sum size – summen af størrelserne på handlerne per minut.
- pvolume – sum price – summen af priserne per minut, nødvendig for avgPrice.
- turnover – sum price*size – den samlede handelsvolumen per minut.
- avgPrice – pvolume%numTrades – gennemsnitlig pris per minut.
- avgSize – volume%numTrades – gennemsnitlig størrelse på handlen per minut.
- vwap – turnover%volume – vægtet gennemsnitlig pris per minut baseret på handlens størrelse.
- cumVolume – sum volume – den samlede størrelse på handlerne over tid.
Lad os straks tage fat på et ikke så indlysende punkt – hvordan vi initierer disse kolonner første gang og for hver efterfølgende minut. Nogle kolonner som firstPrice skal hver gang initieres med værdien null, da deres værdi ikke er defineret. Andre, som volume, skal altid sættes til 0. Der er også kolonner, der kræver en kombineret tilgang – for eksempel skal cumVolume kopieres fra det foregående minut, mens den første skal sættes til 0. Vi vil definere alle disse parametre ved hjælp af datatypen ordbog (analog til optagelse):
// 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 объяснен ниже
Jeg har tilføjet sym og time til ordbogen for nemheds skyld, nu er initWith en klar linje fra den endelige aggregerede tabel, hvor de korrekte sym og time stadig skal angives. Den kan bruges til at tilføje nye rækker til tabellen.
aggCols vil være nødvendige, når vi opretter den aggregerende funktion. Listen skal inverteres på grund af egenskaberne ved rækkefølgen af beregninger i Q (højre mod venstre). Målet er at sikre en beregning i retning fra high til cumVolume, da nogle kolonner afhænger af tidligere værdier.
Kolonner, der skal kopieres til det nye minut fra det forrige, sym kolonnen er tilføjet for nemheds skyld:
rollColumns:`sym`cumVolume;
Nu deler vi kolonnerne op i grupper alt efter, hvordan de skal opdateres. Vi kan identificere tre typer:
- Akkumulatorer (volume, turnover,..) – vi skal lægge den indgående værdi til den tidligere.
- Med særlig punkt (high, low, ..) – den første værdi i minuttet tages fra de indgående data, de øvrige beregnes ved hjælp af funktionen.
- De øvrige. Beregnes altid ved hjælp af funktionen.
Lad os definere variable for disse klasser:
accumulatorCols:`numTrades`volume`pvolume`turnover;
specialCols:`high`low`firstPrice`firstSize;
Beregningens rækkefølge
Vi vil opdatere den aggregerede tabel i to trin. For effektivitet vil vi først komprimere den indgående tabel, så der kun er én række for hver symbol og minut. At alle vores funktioner er inkrementelle og associative garanterer, at resultatet fra dette ekstra skridt ikke vil ændre sig. Tabellen kunne komprimeres ved hjælp af et select:
select high:max price, low:min price … by sym,time.minute from table
Denne metode har en ulempe – sættet af beregnede kolonner er forudbestemt. Heldigvis er select i Q også implementeret som en funktion, hvor der kan indsættes dynamisk oprettede argumenter:
?[table;whereClause;byClause;selectClause]
Jeg vil ikke detaljeret beskrive argumentformatet; i vores tilfælde vil de kun være noget komplekse for by- og select-udtryk, og de skal være ordbøger i form af columns!expressions. Således kan den komprimerende funktion defineres som:
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 er en map-funktion i Q for én liste
preprocess:?[;();`sym`time!`sym`time.minute;selExpression];
For klarhed har jeg brugt funktionen parse, som forvandler en streng med Q-udtryk til en værdi, der kan sendes til funktionen eval og som er nødvendig i en funktionel select. Lad os også bemærke, at preprocess er angivet som en projektion (dvs. en funktion med delvist definerede argumenter) af select-funktionen, ét argument (tabellen) mangler. Hvis vi anvender preprocess på tabellen, får vi en komprimeret tabel.
Det andet trin er opdatering af den aggregerede tabel. Lad os først skrive algoritmen i pseudokode:
for each sym in inputTable
idx: rækkeindeks i agg TABLE for sym+currentTime;
aggTable[idx;`high]: aggTable[idx;`high] | inputTable[sym;`high];
aggTable[idx;`volume]: aggTable[idx;`volume] + inputTable[sym;`volume];
…
I Q er det almindeligt at bruge map/reduce-funktioner i stedet for løkker. Men da Q er et vektorbaseret sprog, og vi kan anvende alle operationer på alle symboler samtidigt, kan vi i første omgang klare os helt uden løkker og udføre operationer på alle symboler på én gang:
idx:calcIdx inputTable;
row:aggTable idx;
aggTable[idx;`high]: row[`high] | inputTable`high;
aggTable[idx;`volume]: row[`volume] + inputTable`volume;
…
Men vi kan gå endnu videre; i Q findes der en unik og meget kraftfuld operator – operatoren for generaliseret tildeling. Denne tillader os at ændre værdien af et komplekst datastruktur ved hjælp af en liste af indekser, funktioner og argumenter. I vores tilfælde ser det sådan ud:
idx:calcIdx inputTable;
rows:aggTable idx;
// .[target;(idx0;idx1;..);function;argument] ~ target[idx 0;idx 1;…]: function[target[idx 0;idx 1;…];argument], i vores tilfælde er funktionen – tildeling
.[aggTable;(idx;aggCols);:;flip (row[`high] | inputTable`high;row[`volume] + inputTable`volume;…)];
Desværre kræver tildeling til tabellen en liste af rækker snarere end kolonner, og det tvinger os til at transponere matricen (fra liste af kolonner til liste af rækker) ved hjælp af funktionen flip. For en stor tabel er dette omkostningsfuldt, så i stedet anvender vi generaliseret tildeling på hver kolonne separat ved hjælp af map-funktionen (som ser ud som et apostrof):
[aggTable;;:;]'[(idx;)each aggCols; (row[`high] | inputTable`high;row[`volume] + inputTable`volume;…)];
Vi bruger igen funktionsprojektion. Bemærk også, at oprettelse af en liste i Q også er en funktion, og vi kan kalde den ved hjælp af funktionen each(map) for at få en liste af lister.
For at sættet af beregnede kolonner ikke skal være fast, vil vi skabe udtrykket ovenfor dynamisk. Vi definerer først funktionerne til at beregne hver kolonne, ved at bruge variablerne row og inp til at referere til de aggregerede og indgående data:
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");
Nogle kolonner er specielle, deres første værdi skal ikke beregnes af funktionen. Vi kan definere, at den er første i kolonnen row[`numTrades] – hvis den er 0, så er værdien første. I Q findes der en valgfunktion — ?[Boolean liste;list1;list2] – som vælger værdi fra liste 1 eller 2 afhængigt af betingelsen i det første argument:
// high -> ?[isFirst;inp`high;row[`high]|inp`high]
// @ - тоже обобщенное присваивание для случая когда индекс неглубокий
@[`aggExpression;specialCols;{[x;y]"?[isFirst;inp`",y,";",x,"]"};string specialCols];
Her har jeg kaldt den generaliserede tildeling med min funktion (udtrykket i krøllede parenteser). Den modtager den nuværende værdi (første argument) og et ekstra argument, som jeg passerer som det 4. parameter.
Tilføjelse af akkumuleringskolonner hver for sig, da funktionen for dem er den samme:
// volume -> row[`volume]+inp`volume
aggExpression[accumulatorCols]:{"row[`",x,"]+inp`",x } each string accumulatorCols;
Dette er en almindelig tildeling efter Q-standarder, men jeg tildeler straks en liste af værdier. Til sidst skaber vi hovedfunktionen:
// ":",/: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),")]}";
Med dette udtryk opretter jeg dynamisk en funktion fra en streng, der indeholder udtrykket, som jeg nævnte ovenfor. Resultatet vil se således ud:
{[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])]}
Beregningen af kolonnerne er inverteret, fordi i Q er rækkefølgen af beregning fra højre mod venstre.
Nu har vi to hovedfunktioner, der kræves for beregningerne, det mangler kun lidt infrastruktur, og så er servicen klar.
Afsluttende trin
Vi har funktionerne preprocess og updateAgg, som gør hele arbejdet. Men vi skal også sikre korrekt overførsel gennem minutter og beregne indeksene til aggregation. Først skal vi definere funktionen init:
init:{
tradeAgg:: 0#enlist[initWith];
currTime::00:00;
currSyms::`u#`symbol$();
offset::0;
rollCache:: `sym xkey update `u#sym from rollColumns#tradeAgg;
}
Vi skal også definere funktionen roll, som vil ændre det aktuelle minut:
roll:{[tm]
if[currTime>tm; :init[]];
rollCache,::offset _ rollColumns#tradeAgg;
offset::count tradeAgg;
currSyms::`u#`$();
}
Vi får brug for en funktion til at tilføje nye symboler:
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)];
}
Og endelig funktionen upd (det traditionelle navn for denne funktion for Q-tjenester), der kaldes af klienten for at tilføje data:
upd:{[tblName;data]
tm:exec distinct time from data:() xkey preprocess data;
updMinute[data] each tm;
};
updMinute:{[data;tm]
if[tmcurrTime; 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];
};
Det var det. Her er den fulde kode til vores tjeneste, som lovet, kun få linjer:
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];
};
Testning
Lad os teste serverens ydeevne. Til dette skal vi køre den i en separat proces (placer koden i filen service.q) og kalde funktionen init:
q service.q –p 5566
q)init[]
I et andet terminalvindue skal du starte en anden Q-proces og forbinde til den første:
h:hopen `:host:5566
h:hopen 5566 // hvis begge er på samme vært
Lad os først oprette en liste over symboler – 10000 stykker, og tilføje en funktion til at generere en tilfældig tabel. I det andet terminalvindue:
syms:`IBM`AAPL`GOOG,-9997?`8
rnd:{[n;t] ([] sym:n?syms; time:t+asc n#til 25; price:n?10f; size:n?10)}
Jeg har tilføjet tre rigtige symboler til symbolerne for at gøre det lettere at finde dem i tabellen. Funktionen rnd genererer en tilfældig tabel med n rækker, hvor tiden varierer fra t til t+25 millisekunder.
Nu kan vi prøve at sende data til serveren (tilføjer de første ti timer):
{h (`upd;`trade;rnd[10000;x])} each `time$00:00 + til 60*10
Du kan kontrollere på serveren, at tabellen er opdateret:
c 25 200
select from tradeAgg where sym=`AAPL
-20#select from tradeAgg where sym=`AAPL
Resultat:
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|2919Lad os nu udføre belastningstest for at finde ud af, hvor mange data tjenesten kan håndtere pr. minut. Jeg minder om, at vi har sat opdateringsintervallet til 25 millisekunder. Tjenesten skal derfor (i gennemsnit) kunne holde sig inden for mindst 20 millisekunder pr. opdatering for at give brugerne tid til at anmode om data. Indtast følgende i den anden proces:
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 – det er to minutter. Man kan starte med 1000 linjer hver 25. millisekund:
start 1000
I mit tilfælde får jeg et resultat på omkring et par millisekunder pr. opdatering. Jeg vil derfor straks øge antallet af linjer til 10.000:
start 10000
Resultat:
min| 00:00:00.004
avg| 9.191458
med| 9f
max| 00:00:00.030
Igen ikke noget særligt, og det er jo 24 millioner linjer pr. minut, 400.000 pr. sekund. Mere end 25 millisekunder opdatering hæmmede kun 5 gange, sandsynligvis ved skift af minut. Lad os øge til 100.000:
start 100000
Resultat:
min| 00:00:00.013
avg| 25.11083
med| 24f
max| 00:00:00.108
q)sum times
00:02:00.532
Som vi kan se, kæmper tjenesten lige præcis, men den formår stadig at holde sig flydende. Sådanne datamængder (240 millioner linjer pr. minut) er ekstremt store, og i sådanne tilfælde er det sædvanligt at køre flere kloner (eller endda dusinvis af kloner) af tjenesten, hvor hver kun behandler en del af symbolerne. Alligevel er resultatet imponerende for et fortolket sprog, der primært er fokuseret på datalagring.
Der kan opstå spørgsmål om, hvorfor tiden vokser ikke-lineært sammen med størrelsen af hver opdatering. Årsagen er, at komprimeringsfunktionen reelt er en C-funktion, der fungerer langt mere effektivt end updateAgg. Fra en vis opdateringsstørrelse (omkring 10.000), når updateAgg sit loft, og dens udførelsestid afhænger ikke længere af opdateringsstørrelsen. Netop på grund af det præliminære Q-trin er tjenesten i stand til at håndtere sådanne datamængder. Dette understreger, hvor vigtigt det er at vælge den rigtige algoritme, når man arbejder med store data. Endnu et punkt er riktig datalagring i hukommelsen. Hvis dataene ikke blev gemt kolonnenært eller ikke var tidsordnet, ville vi have at gøre med et fænomen kaldet TLB cache miss – mangel på sidens adresse i CPU’ens adressecache. At finde adressen tager omkring 30 gange længere tid, hvis det mislykkes, og i tilfælde af spredte data kan det bremse tjenesten med mange gange.
Konklusion
I denne artikel har jeg vist, at KDB+ databasen og Q ikke kun er velegnede til opbevaring af store data og nem adgang til dem via SELECT, men også til at oprette databehandlingsservices, der kan håndtere hundreder af millioner af rækker/ gigabyte data selv i en enkelt Q-proces. Selve Q-sproget muliggør ekstremt kortfattet og effektiv implementering af algoritmer relateret til databehandling takket være sin vektornatur, indbyggede SQL-dialekt interpreter og en meget vellykket samling af bibliotekfunktioner.
Jeg vil bemærke, at det ovenstående kun er en del af Q's muligheder; det har også andre unikke funktioner. For eksempel en ekstremt enkel IPC-protokol, der overskriver grænsen mellem separate Q-processer og muliggør at samle hundreder af disse processer i et enkelt netværk, der kan være lokaliseret på snesevis af servere i forskellige dele af verden.
Kilde: habr.com
