Podeu llegir sobre quina és la base KDB+, el llenguatge de programació Q, quins són els seus punts forts i febles al meu anterior i breument a la introducció. A l'article, implementarem un servei a Q que processarà el flux de dades entrants i calcularà diverses funcions d'agregació cada minut en mode "temps real" (és a dir, tindrà temps per calcular-ho tot abans de la següent porció de dades). La característica principal de Q és que és un llenguatge vectorial que permet operar no amb objectes únics, sinó amb les seves matrius, matrius de matrius i altres objectes complexos. Idiomes com Q i els seus parents K, J, APL són famosos per la seva brevetat. Sovint, un programa que ocupa diverses pantalles de codi en un llenguatge conegut com Java es pot escriure en poques línies. Això és el que vull demostrar en aquest article.
Introducció
KDB+ és una base de dades columnar centrada en quantitats molt grans de dades, ordenades d'una manera específica (principalment per temps). S'utilitza principalment en institucions financeres: bancs, fons d'inversió, companyies d'assegurances. El llenguatge Q és el llenguatge intern de KDB+ que us permet treballar amb eficàcia amb aquestes dades. La ideologia Q és la brevetat i l'eficiència, mentre que la claredat es sacrifica. Això es justifica pel fet que el llenguatge vectorial serà difícil d'entendre en qualsevol cas, i la brevetat i la riquesa de l'enregistrament permet veure una part molt més gran del programa en una pantalla, cosa que en última instància en facilita la comprensió.
En aquest article implementem un programa complet a Q i potser voldreu provar-lo. Per fer-ho, necessitareu la Q real. Podeu descarregar la versió gratuïta de 32 bits al lloc web de l'empresa kx: . Allà, si us interessa, hi trobareu informació de referència sobre Q, el llibre i diversos articles sobre aquest tema.
Declaració de problemes
Hi ha una font que envia una taula amb dades cada 25 mil·lisegons. Com que KDB+ s'utilitza principalment en finances, suposarem que es tracta d'una taula de transaccions (operacions), que té les columnes següents: temps (temps en mil·lisegons), sym (designació de l'empresa a la borsa - IBM, AAPL,…), preu (el preu al qual s'han comprat les accions), mida (mida de l'operació). L'interval de 25 mil·lisegons és arbitrari, ni massa petit ni massa llarg. La seva presència fa que les dades arribin al servei ja en buffer. Seria fàcil implementar la memòria intermèdia al costat del servei, inclosa la memòria intermèdia dinàmica en funció de la càrrega actual, però per simplificar, ens centrarem en un interval fix.
El servei ha de comptar cada minut per a cada símbol entrant de la columna sym un conjunt de funcions d'agregació: preu màxim, preu mitjà, mida de la suma, etc. Informació útil. Per simplificar, assumirem que totes les funcions es poden calcular de manera incremental, és a dir. per obtenir un valor nou, n'hi ha prou de conèixer dos nombres: el valor antic i el valor entrant. Per exemple, les funcions max, average, sum tenen aquesta propietat, però la funció mediana no.
També assumirem que el flux de dades entrant està ordenat en el temps. Això ens donarà l'oportunitat de treballar només amb l'últim moment. A la pràctica, n'hi ha prou amb poder treballar amb els minuts actuals i anteriors en cas que algunes actualitzacions arribin tard. Per simplificar, no considerarem aquest cas.
Funcions d'agregació
Les funcions d'agregació necessàries s'enumeren a continuació. Vaig agafar-ne el màxim possible per augmentar la càrrega del servei:
- alt - preu màxim - preu màxim per minut.
- baix - preu mínim - preu mínim per minut.
- firstPrice – primer preu – primer preu per minut.
- lastPrice - últim preu - últim preu per minut.
- firstSize: primera mida: primera mida comercial per minut.
- lastSize: última mida: última mida comercial en un minut.
- numTrades – count i – nombre d'operacions per minut.
- volum - mida de la suma - suma de les mides comercials per minut.
- pvolum - preu suma - suma de preus per minut, necessària per a avgPrice.
- - preu de facturació suma * mida - volum total de transaccions per minut.
- avgPrice - pvolume%numTrades - preu mitjà per minut.
- avgSize: volum%numTrades: mida mitjana del comerç per minut.
- vwap - volum de negoci% - preu mitjà per minut ponderat per la mida de la transacció.
- cumVolume: volum suma: mida acumulada de transaccions durant tot el temps.
Parlem immediatament d'un punt no obvi: com inicialitzar aquestes columnes per primera vegada i per a cada minut posterior. Algunes columnes del tipus firstPrice s'han d'inicialitzar a null cada vegada; el seu valor no està definit. Els altres tipus de volum sempre s'han d'establir a 0. També hi ha columnes que requereixen un enfocament combinat; per exemple, cumVolume s'ha de copiar des del minut anterior i, per al primer, s'ha de posar a 0. Establim tots aquests paràmetres utilitzant les dades del diccionari. tipus (analògic a un registre):
// 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 объяснен ниже
He afegit sym i time al diccionari per comoditat, ara initWith és una línia ja feta de la taula agregada final, on queda establir el sym i l'hora correctes. Podeu utilitzar-lo per afegir noves files a una taula.
Necessitarem aggCols quan creem una funció d'agregació. La llista s'ha d'invertir a causa de l'ordre en què s'avaluen les expressions de Q (de dreta a esquerra). L'objectiu és assegurar-se que el càlcul va d'alt a cumVolume, ja que algunes columnes depenen de les anteriors.
A les columnes que s'han de copiar a un nou minut de l'anterior, s'afegeix la columna sym per comoditat:
rollColumns:`sym`cumVolume;
Ara dividim les columnes en grups segons com s'han d'actualitzar. Es poden distingir tres tipus:
- Acumuladors (volum, facturació,..) – hem d'afegir el valor d'entrada a l'anterior.
- Amb un punt especial (alt, baix, ..): el primer valor del minut s'agafa de les dades entrants, la resta es calculen mitjançant la funció.
- Descans. Sempre es calcula mitjançant una funció.
Definim variables per a aquestes classes:
accumulatorCols:`numTrades`volume`pvolume`turnover;
specialCols:`high`low`firstPrice`firstSize;
Ordre de càlcul
Actualitzarem la taula agregada en dues etapes. Per eficiència, primer reduïm la taula d'entrada de manera que només hi hagi una fila per cada caràcter i minut. El fet que totes les nostres funcions siguin incrementals i associatives garanteix que el resultat d'aquest pas addicional no canviarà. Podeu reduir la taula amb select:
select high:max price, low:min price … by sym,time.minute from table
Aquest mètode té un desavantatge: el conjunt de columnes calculades està predefinit. Afortunadament, a Q, select també s'implementa com una funció on podeu substituir arguments creats dinàmicament:
?[table;whereClause;byClause;selectClause]
No descriuré amb detall el format dels arguments; en el nostre cas, només les expressions by i select no seran trivials i haurien de ser diccionaris de la forma columnes!expressions. Així, la funció de contracció es pot definir de la següent manera:
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];
Per a més claredat, vaig utilitzar la funció d'anàlisi, que converteix una cadena amb una expressió Q en un valor que es pot passar a la funció d'eval i que es requereix a la funció de selecció. Tingueu en compte també que el preprocés es defineix com una projecció (és a dir, una funció amb arguments parcialment definits) de la funció de selecció, falta un argument (la taula). Si apliquem un preprocés a una taula, obtindrem una taula comprimida.
La segona etapa és l'actualització de la taula agregada. Primer escrivim l'algorisme en pseudocodi:
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, és habitual utilitzar funcions de mapa/reducció en lloc de bucles. Però com que Q és un llenguatge vectorial i podem aplicar fàcilment totes les operacions a tots els símbols alhora, a una primera aproximació podem prescindir d'un bucle, fent operacions sobre tots els símbols alhora:
idx:calcIdx inputTable;
row:aggTable idx;
aggTable[idx;`high]: row[`high] | inputTable`high;
aggTable[idx;`volume]: row[`volume] + inputTable`volume;
…
Però podem anar més enllà, Q té un operador únic i extremadament potent: l'operador d'assignació generalitzada. Permet canviar un conjunt de valors en una estructura de dades complexa mitjançant una llista d'índexs, funcions i arguments. En el nostre cas es veu així:
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;…)];
Malauradament, per assignar a una taula necessiteu una llista de files, no columnes, i heu de transposar la matriu (llista de columnes a llista de files) mitjançant la funció flip. Això és car per a una taula gran, de manera que apliquem una assignació generalitzada a cada columna per separat, utilitzant la funció de mapa (que sembla un apòstrof):
.[aggTable;;:;]'[(idx;)each aggCols; (row[`high] | inputTable`high;row[`volume] + inputTable`volume;…)];
Tornem a utilitzar la projecció de funcions. Tingueu en compte també que a Q, crear una llista també és una funció i la podem cridar utilitzant la funció each(map) per obtenir una llista de llistes.
Per assegurar-nos que el conjunt de columnes calculades no sigui fix, crearem l'expressió anterior de manera dinàmica. Primer definim funcions per calcular cada columna, utilitzant les variables fila i inp per fer referència a les dades agregades i d'entrada:
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");
Algunes columnes són especials; el seu primer valor no hauria de ser calculat per la funció. Podem determinar que és el primer per la columna fila[`numTrades]; si conté 0, el valor és el primer. Q té una funció de selecció - ?[Llista booleana;lista1;lista2] - que selecciona un valor de la llista 1 o 2 depenent de la condició del primer argument:
// high -> ?[isFirst;inp`high;row[`high]|inp`high]
// @ - тоже обобщенное присваивание для случая когда индекс неглубокий
@[`aggExpression;specialCols;{[x;y]"?[isFirst;inp`",y,";",x,"]"};string specialCols];
Aquí vaig trucar una assignació generalitzada amb la meva funció (una expressió entre claus). Rep el valor actual (el primer argument) i un argument addicional, que passo al 4t paràmetre.
Afegim altaveus de bateria per separat, ja que la funció és la mateixa per a ells:
// volume -> row[`volume]+inp`volume
aggExpression[accumulatorCols]:{"row[`",x,"]+inp`",x } each string accumulatorCols;
Aquesta és una assignació normal pels estàndards Q, però estic assignant una llista de valors alhora. Finalment, creem la funció principal:
// ":",/: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),")]}";
Amb aquesta expressió, creo dinàmicament una funció a partir d'una cadena que conté l'expressió que vaig donar més amunt. El resultat serà així:
{[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])]}
L'ordre d'avaluació de la columna s'inverteix perquè a Q l'ordre d'avaluació és de dreta a esquerra.
Ara tenim dues funcions principals necessàries per als càlculs, només cal afegir una mica d'infraestructura i el servei està llest.
Passos finals
Tenim funcions de preprocessament i updateAgg que fan tota la feina. Però encara cal garantir la transició correcta a través dels minuts i calcular els índexs per a l'agregació. Primer de tot, anem a definir la funció 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
}
També definirem la funció roll, que canviarà el minut actual:
roll:{[tm]
if[currTime>tm; :init[]]; // если перевалили за полночь, то просто вызовем init
rollCache,::offset _ rollColumns#tradeAgg; // обновим кэш – взять roll колонки из aggTable, обрезать, вставить в rollCache
offset::count tradeAgg;
currSyms::`u#`$();
}
Necessitarem una funció per afegir nous caràcters:
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)];
}
I, finalment, la funció upd (el nom tradicional d'aquesta funció per als serveis Q), que el client crida per afegir dades:
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]; // обновим агрегированную таблицу. Функция ? ищет индекс элементов списка справа в списке слева.
};
Això és tot. Aquí teniu el codi complet del nostre servei, tal com s'havia promès, només unes poques línies:
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];
};
Proves
Comprovem el rendiment del servei. Per fer-ho, anem a executar-lo en un procés separat (posar el codi al fitxer service.q) i cridar a la funció init:
q service.q –p 5566
q)init[]
En una altra consola, inicieu el segon procés Q i connecteu-vos al primer:
h:hopen `:host:5566
h:hopen 5566 // если оба на одном хосте
Primer, creem una llista de símbols: 10000 peces i afegim una funció per crear una taula aleatòria. A la segona consola:
syms:`IBM`AAPL`GOOG,-9997?`8
rnd:{[n;t] ([] sym:n?syms; time:t+asc n#til 25; price:n?10f; size:n?10)}
He afegit tres símbols reals a la llista perquè sigui més fàcil cercar-los a la taula. La funció rnd crea una taula aleatòria amb n files, on el temps varia de t a t+25 mil·lisegons.
Ara podeu provar d'enviar dades al servei (afegiu les deu primeres hores):
{h (`upd;`trade;rnd[10000;x])} each `time$00:00 + til 60*10
Podeu comprovar al servei que la taula s'ha actualitzat:
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|2919Ara realitzem proves de càrrega per esbrinar quantes dades pot processar el servei per minut. Us recordo que vam establir l'interval d'actualització en 25 mil·lisegons. En conseqüència, el servei ha d'encaixar (de mitjana) almenys 20 mil·lisegons per actualització per donar temps als usuaris per demanar dades. Introduïu el següent en el segon procés:
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 són dos minuts. Podeu provar d'executar primer durant 1000 files cada 25 mil·lisegons:
start 1000
En el meu cas, el resultat és d'uns mil·lisegons per actualització. Així que immediatament augmentaré el nombre de files a 10.000:
start 10000
Resultat:
min| 00:00:00.004
avg| 9.191458
med| 9f
max| 00:00:00.030
De nou, res especial, però això són 24 milions de línies per minut, 400 mil per segon. Durant més de 25 mil·lisegons, l'actualització es va alentir només 5 vegades, pel que sembla quan va canviar el minut. Augmentem a 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
Com podeu veure, el servei amb prou feines pot fer front, però, tanmateix, aconsegueix mantenir-se a flotació. Aquest volum de dades (240 milions de files per minut) és extremadament gran; en aquests casos, és habitual llançar diversos clons (o fins i tot desenes de clons) del servei, cadascun dels quals processa només una part dels personatges. Tot i així, el resultat és impressionant per a un llenguatge interpretat que se centra principalment en l'emmagatzematge de dades.
Pot sorgir la pregunta de per què el temps creix de manera no lineal amb la mida de cada actualització. El motiu és que la funció de reducció és en realitat una funció C, que és molt més eficient que updateAgg. A partir d'una determinada mida d'actualització (al voltant de 10.000), updateAgg arriba al sostre i aleshores el seu temps d'execució no depèn de la mida de l'actualització. És a causa del pas preliminar Q que el servei és capaç de digerir aquests volums de dades. Això posa de manifest l'important que és triar l'algoritme adequat quan es treballa amb big data. Un altre punt és l'emmagatzematge correcte de les dades a la memòria. Si les dades no s'emmagatzemen en columna o no s'ordenen per temps, ens familiaritzaríem amb una fallada de memòria cau TLB: l'absència d'una adreça de pàgina de memòria a la memòria cau d'adreces del processador. La cerca d'una adreça triga unes 30 vegades més si no té èxit, i si les dades estan disperses, pot alentir el servei diverses vegades.
Conclusió
En aquest article, vaig demostrar que la base de dades KDB+ i Q són adequades no només per emmagatzemar dades grans i accedir-hi fàcilment mitjançant la selecció, sinó també per crear serveis de processament de dades capaços de digerir centenars de milions de files/gigabytes de dades fins i tot en un sol procés Q. El propi llenguatge Q permet una implementació extremadament concisa i eficient d'algorismes relacionats amb el processament de dades a causa de la seva naturalesa vectorial, intèrpret de dialectes SQL integrat i un conjunt de funcions de biblioteca molt reeixit.
Tindré en compte que l'anterior és només una part del que pot fer Q, també té altres característiques úniques. Per exemple, un protocol IPC extremadament senzill que esborra el límit entre els processos Q individuals i permet combinar centenars d'aquests processos en una única xarxa, que es pot localitzar en desenes de servidors de diferents parts del món.
Font: www.habr.com