Om vad KDB+ är, programmeringsspråket Q, och vilka styrkor och svagheter de har kan du läsa i min tidigare och kort i introduktionen. I denna artikel kommer vi att implementera en tjänst på Q som kommer att bearbeta inkommande datastream och beräkna olika aggregatfunktioner per minut i "real-tid" (d.v.s. den kommer att hinna räkna allt innan nästa datamängd kommer). Huvudfunktionen hos Q är att det är ett vektorbaserat språk som tillåter oss att arbeta med inte bara enskilda objekt utan även med deras arrayer, arrayer av arrayer och andra komplexa sammansatta objekt. Språk som Q och dess släktingar K, J, APL är kända för sin kortfattade syntax. Ofta kan ett program som tar flera skärmar kod i vanliga språk som Java skrivas i ett par rader med dem. Det är precis det jag vill visa i denna artikel.
Introduktion
KDB+ är en kolumnbaserad databas som är inriktad på mycket stora datamängder som är sorterade på ett visst sätt (främst efter tid). Den används främst i finansorganisationer – banker, investeringsfonder, försäkringsbolag. Språket Q är det interna språket i KDB+ som möjliggör effektiv hantering av dessa data. Ideologin bakom Q är korthet och effektivitet, där tydlighet ibland offras. Detta motiveras med att ett vektorbaserat språk alltid kommer att vara svårt att ta till sig, och kortheten och kompaktheten i koden gör att man kan se en mycket större del av programmet på en skärm, vilket underlättar förståelsen.
I artikeln kommer vi att implementera ett fullt fungerande program i Q och du kanske vill prova det i praktiken. För detta behöver du Q. Du kan ladda ner en gratis 32-bitars version från kx:s hemsida – . Där hittar du också, om du är intresserad, referensinformation om Q, boken , och olika artiklar om ämnet.
Uppgiftsställande
Det finns en källa som skickar en datatabell var 25:e millisekund. Eftersom KDB+ främst används inom finans kommer vi att anta att detta är en handelstabell (trades) med följande kolumner: time (tid i millisekunder), sym (börskod för bolag – IBM, AAPL,…), price (pris, till vilket aktierna köptes), size (storleken på affären). Intervallet på 25 millisekunder är valt godtyckligt, det är varken för litet eller för stort. Dess närvaro betyder att data kommer till tjänsten redan buffrat. Det skulle vara lätt att genomföra buffring på tjänstens sida, även dynamisk, beroende på nuvarande belastning, men för enkelhetens skull kommer vi att stanna vid ett fast intervall.
Tjänsten bör beräkna per minut för varje inkommande tecken i kolumnen sym en uppsättning aggregateringsfunktioner – max price, avg price, sum size och så vidare som är användbar information. För enkelhetens skull antar vi att alla funktioner kan beräknas inkrementellt, dvs. för att få ett nytt värde behöver vi bara veta två tal – det gamla och det inkommande värdet. Till exempel har funktionerna max, average, sum denna egenskap, medan medianfunktionen inte har det.
Vi förutsätter också att den inkommande datastreamen är tidsordnad. Detta ger oss möjlighet att endast arbeta med den senaste minuten. I praktiken är det tillräckligt att kunna arbeta med den aktuella och föregående minuten ifall vissa uppdateringar är försenade. För enkelhetens skull kommer vi inte att överväga detta fall.
Aggregationsfunktioner
Nedan listas de nödvändiga aggregationsfunktionerna. Jag har tagit så många som möjligt för att öka belastningen på tjänsten:
- high – max price – maximalt pris per minut.
- low – min price – minimalt pris per minut.
- firstPrice – first price – första pris per minut.
- lastPrice – last price – sista pris per minut.
- firstSize – first size – första storlek på affären per minut.
- lastSize – last size – sista storlek på affären per minut.
- numTrades – count i – antal affärer per minut.
- volume – sum size – summa av storlekar på affärer per minut.
- pvolume – sum price – summa priser per minut, nödvändig för avgPrice.
- turnover – sum price*size – totalt affärsvärde per minut.
- avgPrice – pvolume%numTrades – genomsnittligt pris per minut.
- avgSize – volume%numTrades – genomsnittlig storlek på affären per minut.
- vwap – turnover%volume – vägt genomsnittligt pris per minut baserat på affärsstorlek.
- cumVolume – sum volume – ackumulerad storlek på affärer genom alla tider.
Låt oss direkt diskutera en subtil punkt – hur man initierar dessa kolumner första gången och för varje följande minut. Vissa kolumner som firstPrice måste alltid initieras med värdet null eftersom deras värde inte är definierat. Andra som volume bör alltid sättas till 0. Det finns också kolumner som kräver en kombinerad metod – till exempel måste cumVolume kopieras från föregående minut, och för den första sättas till 0. Vi ställer in alla dessa parametrar med hjälp av datatypen dictionary (analog till struktur):
// 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 объяснен ниже
Jag har lagt till sym och time i ordboken för bekvämlighet, nu är initWith en färdig rad från den slutgiltiga aggregerade tabellen, där vi bara behöver ange rätt sym och time. Den kan användas för att lägga till nya rader i tabellen.
aggCols kommer vi att behöva vid skapandet av den aggregerande funktionen. Listan måste inverteras på grund av egenskaperna hos beräkningsordningen för uttryck i Q (höger till vänster). Målet är att säkerställa beräkning i riktning från high till cumVolume, eftersom vissa kolumner är beroende av föregående.
Kolumner som behöver kopieras till den nya minuten från den föregående, kolumnen sym har lagts till för bekvämlighet:
rollColumns:`sym`cumVolume;
Nu kommer vi att dela upp kolumnerna i grupper enligt hur de ska uppdateras. Vi kan dela in dem i tre typer:
- Ackumulatorer (volume, turnover,..) – vi måste lägga till det inkommande värdet till det föregående.
- Med särskilt värde (high, low, ..) – det första värdet på minuten tas från de inkommande uppgifterna, de andra beräknas med hjälp av en funktion.
- Övriga. Beräknas alltid med hjälp av en funktion.
Låt oss definiera variabler för dessa klasser:
accumulatorCols:`numTrades`volume`pvolume`turnover;
specialCols:`high`low`firstPrice`firstSize;
Beräkningsordning
Vi kommer att uppdatera den aggregerade tabellen i två steg. För effektivitet kommer vi först att komprimera den inkommande tabellen så att det bara finns en rad för varje symbol och minut. Att alla våra funktioner är inkrementella och associativa garanterar att resultatet av detta extra steg inte kommer att ändras. Tabellen skulle kunna komprimeras med hjälp av en selektion:
select high:max price, low:min price … by sym,time.minute from table
Nackdelen med denna metod är att mängden beräknade kolumner anges i förväg. Lyckligtvis har Q selekt även implementerats som en funktion, där man kan sätta in dynamiskt skapade argument:
?[table;whereClause;byClause;selectClause]
Jag kommer inte att gå in på detaljer om argumentformatet, i vårt fall kommer endast by- och select-uttrycken att vara icke-triviala och de måste vara ordböcker av typen columns!expressions. På så sätt kan vi ange komprimeringsfunktionen så här:
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 är en map-funktion i Q för en lista
preprocess:?[;();`sym`time!`sym`time.minute;selExpression];
För tydlighetens skull använde jag funktionen parse, som omvandlar en sträng med Q-uttryck till ett värde som kan skickas till funktionen eval och som krävs i det funktionella valet. Vi noterar också att preprocess anges som en projektion (d.v.s. en funktion med delvis definierade argument) av select-funktionen, en argument (tabell) saknas. Om vi applicerar preprocess på tabellen får vi en komprimerad tabell.
Det andra steget är att uppdatera den aggregerade tabellen. Vi börjar med att skriva algoritmen i pseudokod:
for each sym in inputTable
idx: radindex i agg-tabellen för sym+currentTime;
aggTable[idx;`high]: aggTable[idx;`high] | inputTable[sym;`high];
aggTable[idx;`volume]: aggTable[idx;`volume] + inputTable[sym;`volume];
…
I Q är det vanligt att använda map/reduce-funktioner istället för loopar. Men eftersom Q är ett vektorspråk och vi kan tillämpa alla operationer på alla symboler samtidigt, så kan vi i första hand klara oss helt utan loopar och utföra operationerna på alla symboler på en gång:
idx:calcIdx inputTable;
row:aggTable idx;
aggTable[idx;`high]: row[`high] | inputTable`high;
aggTable[idx;`volume]: row[`volume] + inputTable`volume;
…
Men vi kan gå ännu längre, i Q finns det en unik och extremt kraftfull operator – operatorn för generaliserad tilldelning. Den gör det möjligt att ändra en uppsättning värden i en komplex datastruktur med hjälp av en lista av index, funktioner och argument. I vårt fall ser den ut så här:
idx:calcIdx inputTable;
rows:aggTable idx;
// .[target;(idx0;idx1;..);function;argument] ~ target[idx 0;idx 1;…]: function[target[idx 0;idx 1;…];argument], i vårt fall är funktionen tilldelning
.[aggTable;(idx;aggCols);:;flip (row[`high] | inputTable`high;row[`volume] + inputTable`volume;…)];
Tyvärr krävs en lista med rader för att tilldela i tabellen, inte kolumner, och vi måste transponera matrisen (en lista av kolumner till en lista med rader) med hjälp av funktionen flip. För en stor tabell är detta kostsamt, så istället tillämpar vi generaliserad tilldelning på varje kolumn separat med hjälp av funktionen map (som ser ut som ett apostrof):
[aggTable;;:;]'[(idx;)each aggCols; (row[`high] | inputTable`high;row[`volume] + inputTable`volume;…)]
Vi använder återigen en funktionsprojektion. Observera också att i Q att skapa en lista också är en funktion och vi kan anropa den med hjälp av funktionen each(map) för att få en lista av listor.
För att settet av beräknade kolumner inte ska vara fast, kommer vi att skapa uttrycket ovan dynamiskt. Vi definierar först funktionerna för att beräkna varje kolumn, med hjälp av variablerna row och inp för att referera till de aggregerade och ingående uppgifterna:
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");
Vissa kolumner är speciella, deras första värde ska inte beräknas av funktionen. Vi kan fastställa att det är första genom kolumnen row[`numTrades] – om den har värdet 0, så är värdet det första. I Q finns en valfunktion — ?[Boolean list;list1;list2] — som väljer ett värde från lista 1 eller 2 beroende på villkoret i den första argumentet:
// high -> ?[isFirst;inp`high;row[`high]|inp`high]
// @ - тоже обобщенное присваивание для случая когда индекс неглубокий
@[`aggExpression;specialCols;{[x;y]"?[isFirst;inp`",y,";",x,"]"};string specialCols];
Här har jag anropat en generell tilldelning med min funktion (uttrycket i klamrar). Den tar det aktuella värdet (första argumentet) och ett extra argument som jag skickar som den 4:e parametern.
Vi lägger till ackumulatorkolumner separat, eftersom de har samma funktion:
// volume -> row[`volume]+inp`volume
aggExpression[accumulatorCols]:{"row[`",x,"]+inp`",x } each string accumulatorCols;
Detta är en vanlig tilldelning enligt Q; jag tilldelar direkt en lista med värden. Slutligen kommer vi att skapa huvudfunktionen:
// ":",/: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 detta uttryck skapar jag dynamiskt en funktion från en sträng som innehåller det uttryck jag nämnde tidigare. Resultatet kommer att se ut så här:
{[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])]}
Ordningen för beräkningarna av kolumnerna är omvänd, eftersom i Q beräknas ordningen höger till vänster.
Nu har vi två huvudfunktioner som behövs för beräkningarna, det återstår bara att lägga till lite infrastruktur och tjänsten är klar.
Sista stegen
Vi har funktionerna preprocess och updateAgg, som gör allt arbete. Men vi behöver också säkerställa rätt övergång genom minuterna och beräkna index för aggregeringen. Först bestämmer vi funktionen init:
init:{
tradeAgg:: 0#enlist[initWith]; // skapar en tom typad tabell, enlist omvandlar ordboken till tabell, och 0# betyder att ta 0 element från den
currTime::00:00; // börjar med 0, :: betyder att tilldelningen går till en global variabel
currSyms::`u#`symbol$(); // `u# - omvandlar listan till ett träd för att snabba upp sökningen av element
offset::0; // index i tradeAgg där nuvarande minut börjar
rollCache:: `sym xkey update `u#sym from rollColumns#tradeAgg; // cache för de senaste värdena i roll kolumnerna, tabell med nyckeln sym
}
Vi ska också definiera en roll-funktion som ändrar den aktuella minuten:
roll:{[tm]
if[currTime>tm; :init[]]; // om vi har passerat midnatt, kalla bara på init
rollCache,::offset _ rollColumns#tradeAgg; // uppdatera cachen – hämta roll kolumner från aggTable, klipp, sätt in i rollCache
offset::count tradeAgg;
currSyms::`u#`$();
}
Vi behöver en funktion för att lägga till nya symboler:
addSyms:{[syms]
currSyms,::syms; // lägg till i listan över kända
// lägg till i tabellen sym, time och rollColumns genom att använda generell tilldelning.
// Funktionen ^ sätter standardvärden för roll kolumner om symbolen inte finns i cachen. value flip table returnerar en lista över kolumner i tabellen.
`tradeAgg upsert @[count[syms]#enlist initWith;`sym`time,cols rc;:;(syms;currTime), (initWith cols rc)^value flip rc:rollCache ([] sym: syms)];
}
Och slutligen, funktionen upd (det traditionella namnet för denna funktion för Q-tjänster), som anropas av klienten för att lägga till data:
upd:{[tblName;data] // tblName är inte nödvändig, men vanligtvis hanterar tjänsten flera tabeller
tm:exec distinct time from data:() xkey preprocess data; // preprocess & calc time
updMinute[data] each tm; // lägg till data för varje minut
};
updMinute:{[data;tm]
if[tmcurrTime; roll tm; currTime::tm]; // ändra minut om det behövs
data:select from data where time=tm; // filtrering
if[count msyms:syms where not (syms:data`sym)in currSyms; addSyms msyms]; // nya symboler
updateAgg[`tradeAgg;offset+currSyms?syms;data]; // uppdatera den aggregerade tabellen. Funktionen ? söker efter index för element i listan till höger i listan till vänster.
};
Så, det är allt. Här är hela koden för vår tjänst, som utlovat, bara några rader:
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
Låt oss kontrollera tjänstens prestanda. För detta syfte startar vi den i en separat process (lägg koden i filen service.q) och anropar funktionen init:
q service.q –p 5566
q)init[]
I en annan konsol, starta den andra Q-processen och koppla upp dig mot den första:
h:hopen `:host:5566
h:hopen 5566 // om båda är på samma värd
Först skapar vi en lista med symboler – 10000 stycken och lägger till en funktion för att skapa en slumpmässig tabell. I den andra konsolen:
syms:`IBM`AAPL`GOOG,-9997?`8
rnd:{[n;t] ([] sym:n?syms; time:t+asc n#til 25; price:n?10f; size:n?10)}
Jag har lagt till tre riktiga symboler i listan för att göra dem lättare att hitta i tabellen. Funktionen rnd skapar en slumpmässig tabell med n rader, där tiden ändras från t till t+25 millisekunder.
Nu kan vi försöka skicka data till tjänsten (vi lägger till de första tio timmarna):
{h (`upd;`trade;rnd[10000;x])} each `time$00:00 + til 60*10
Du kan kontrollera i tjänsten att tabellen har uppdaterats:
c 25 200
select from tradeAgg where sym=`AAPL
-20#select from tradeAgg where sym=`AAPL
Resultatet:
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|2919Nu ska vi genomföra belastningstestning för att ta reda på hur mycket data tjänsten kan hantera per minut. Jag påminner om att vi har ställt in en uppdateringsintervall på 25 millisekunder. Därför bör tjänsten (i genomsnitt) klara av att utföra uppdateringen på minst 20 millisekunder för att ge tid åt användarna att begära data. Skriv följande i den andra processen:
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 är två minuter. Vi kan försöka köra först för 1000 rader var 25 millisekund:
start 1000
I mitt fall resulterar det i ett par millisekunder per uppdatering. Så jag ökar antalet rader till 10 000 direkt:
start 10000
Resultatet:
min| 00:00:00.004
avg| 9.191458
med| 9f
max| 00:00:00.030
Återigen ingenting speciellt, trots att det handlar om 24 miljoner rader per minut, 400 000 per sekund. Mer än 25 millisekunder för uppdateringen inträffade bara 5 gånger, tydligen vid byte av minut. Låt oss öka till 100 000:
start 100000
Resultatet:
min| 00:00:00.013
avg| 25.11083
med| 24f
max| 00:00:00.108
q)sum times
00:02:00.532
Som vi ser, klarar tjänsten sig precis, men lyckas ändå hålla sig flytande. En sådan datavolym (240 miljoner rader per minut) är extremt stor, i sådana fall är det vanligt att köra flera kloner (eller till och med dussintals kloner) av tjänsten, var och en av dem som bara hanterar en del av symbolerna. Trots det är resultatet imponerande för ett tolkande språk som främst fokuserar på datalagring.
Det kan uppstå en fråga om varför tiden växer icke-linjärt med storleken på varje uppdatering. Anledningen är att komprimeringsfunktionen faktiskt är en C-funktion som arbetar mycket mer effektivt än updateAgg. Från en viss storlek av uppdateringen (någonstans runt 10 000) når updateAgg sin gräns och dess exekveringstid beror inte längre på storleken på uppdateringen. Just genom det preliminära steget Q kan tjänsten hantera sådana datamängder. Detta understryker hur viktigt det är att välja rätt algoritm när man arbetar med stora data. En annan aspekt är korrekt lagring av data i minnet. Om data inte lagrades kolumnvis eller inte var tidsordnade, skulle vi få lära känna en sak som TLB cache miss – avsaknad av en minnesadress i processorns adresscache. Att söka efter adressen tar ungefär 30 gånger längre tid i händelse av miss och spridda data kan sakta ner tjänsten flera gånger.
Slutsats
I den här artikeln har jag visat att KDB+-databasen och Q är lämpliga inte bara för att lagra stora data och enkelt få åtkomst till dem via SELECT, utan också för att skapa databehandlingstjänster som kan hantera hundratals miljoner rader/gigabyte data även i en enda Q-process. Språket Q möjliggör exceptionellt kort och effektiv implementering av algoritmer relaterade till databehandling tack vare dess vektoriska natur, inbyggda SQL-dialekt och ett mycket välfunnet bibliotek av funktioner.
Jag vill påpeka att det som beskrivs ovan bara är en del av Q:s möjligheter; det har också andra unika egenskaper. Till exempel en extremt enkel IPC-protokoll som suddar ut gränsen mellan separata Q-processer och gör det möjligt att kombinera hundratals av dessa processer i ett enhetligt nätverk som kan sträcka sig över tiotals servrar på olika platser i världen.
Källa: habr.com
