Възможности на езика Q и KDB+ чрез примера на услуга в реално време

За това какво представлява базата KDB+, езика за програмиране Q, техните силни и слаби страни, можете да прочетете в предишната ми статията и кратко във введението. В тази статия ще реализираме услуга на Q, която ще обработва входящ поток от данни и ще изчислява на минутна база различни агрегатни функции в режим на "реално време" (т.е. ще успява да изчисли всичко преди следващата порция данни). Основната особеност на Q е, че това е векторен език, позволяващ работа не само с единични обекти, а с масиви, масиви от масиви и други сложни структури. Языци подобни на Q, като K, J, APL са известни със своето краткост. Често програма, която заема няколко екрана код на познат език като Java, може да бъде написана на тях за няколко реда. Именно това и искам да демонстрирам в тази статия.

Въведение

KDB+ е колоночна база данни, ориентирана към много големи обеми данни, организирани по определен начин (в най-голяма степен по време). Тя се използва предимно във финансови организации – банки, инвестиционни фондове, застрахователни компании. Езикът Q е вътрешният език на KDB+, който позволява ефективна работа с тези данни. Идеологията на Q е краткост и ефективност, а разбирането е жертвано. Това се обосновава с факта, че векторният език така или иначе ще бъде сложен за възприемане, а краткостта на записа позволява да видите на един екран много по-голяма част от програмата, което в крайна сметка улеснява нейното разбиране.

В статията ще реализираме пълноценна програма на Q и вие, вероятно, ще искате да я изпробвате на практика. За целта ще ви е нужен самият Q. Можете да изтеглите безплатната 32-битна версия от сайта на компанията kx – www.kx.com. Там ще намерите и справочна информация за Q, книгата Q For Mortals и разнообразни статии по темата.

Формулиране на задачата

Има източник, който изпраща таблица с данни на всеки 25 милисекунди. Понеже KDB+ се прилага предимно във финансите, да приемем, че това е таблица на сделки (trades), в която има следните колони: time (време в милисекунди), sym (обозначение на компанията на борсата – IBM, AAPL,…), цена (цената, по която са купени акциите), размер (размер на сделката). Интервал от 25 милисекунди е избран произволно, той не е твърде малък и не е твърде голям. Неговото наличие означава, че данните влизат в услугата вече буферирани. Лесно можеше да се реализира буфериране от страна на услугата, включително динамично, в зависимост от текущото натоварване, но за простота ще се придържаме към фиксиран интервал.

Услугата трябва да изчислява поминутно за всеки входящ символ от колоната sym набор от агрегиращи функции - max price, avg price, sum size и т.н. полезна информация. За простота ще предположим, че всички функции могат да се изчисляват инкрементално, т.е. за получаване на нова стойност е достатъчно да знаем две числа - старата и входящата стойност. Например, функциите max, average, sum притежават това свойство, но функцията медиана не.

Също така ще предположим, че входящият поток от данни е подреден по време. Това ни дава възможност да работим само с последната минута. На практика е достатъчно да можем да работим с текущата и предходната минута, в случай че някои актуализации закъсняват. За простота няма да разглеждаме този случай.

Агрегиращи функции

По-долу са изброени необходимите агрегиращи функции. Взех ги колкото е възможно повече, за да увелича натоварването на услугата:

• high – max price – максималната цена за минута.
• low – min price – минималната цена за минута.
• firstPrice – first price – първата цена за минута.
• lastPrice – last price – последната цена за минута.
• firstSize – first size – първият размер на сделката за минута.
• lastSize – last size — последният размер на сделката за минута.
• numTrades – count i – числото на сделките за минута.
• volume – sum size – сумата на размерите на сделките за минута.
• pvolume – sum price – сумата на цените за минута, необходима за avgPrice.
• turnover – sum price*size – сумарният обем на сделките за минута.
• avgPrice – pvolume%numTrades – средната цена за минута.
• avgSize – volume%numTrades – средният размер на сделката за минута.
• vwap – turnover%volume – взетата в предвид средна цена за минута, претеглена по размера на сделките.
• cumVolume – sum volume – натрупаният размер на сделките за всеки период.

Нека да обсъдим един неочевиден момент – как да инициализираме тези колони за първи път и за всяка следваща минута. Някои колони, като firstPrice, трябва всеки път да се инициализират със стойност null, техните стойности не са определени. Други колони, като volume, трябва винаги да се задават на 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 и time в речника за удобство, сега initWith е готовият ред от финалната агрегирана таблица, където остава само да зададем правилните sym и time. Може да се използва за добавяне на нови редове в таблицата.

aggCols ще ни е нужен при създаването на агрегираща функция. Списъкът трябва да се обърне заради особеностите на реда на изчисление на изразите в Q (отдясно наляво). Целта е да се осигури изчисление в посока от high към cumVolume, тъй като някои колони зависят от предишните.

Колоните, които трябва да се копират в новата минута от предишната, колоната sym е добавена за удобство:

rollColumns:`sym`cumVolume;

Сега ще разделим колоните на групи в зависимост от начина, по който трябва да се актуализират. Можем да определим три типа:

1. Акумулатори (volume, turnover,..) – трябва да добавим входящата стойност към предишната.
2. С особена точка (high, low, ..) – първата стойност в минутата се взима от входящите данни, останалите се изчисляват с помощта на функция.
3. Останалите. Винаги се изчисляват с помощта на функция.

Определяме променливи за тези класове:

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]

Няма да описвам подробно формата на аргументите, в нашия случай нетривиални ще бъдат само by и select изрази и те трябва да са речници от вида columns!expressions. По този начин, компресионната функция може да се зададе така:

selExpression:`high`low`firstPrice`lastPrice`firstSize`lastSize`numTrades`volume`pvolume`turnover!parse each ("макс. цена";"мин. цена";"първа цена";"последна цена";"първи размер";"последен размер";"броя i";"сума размер";"сума цена";"сума цена*размер"); // each е функция map в Q за един списък
preprocess:?[;();`sym`time!`sym`time.minute;selExpression];

За да бъде ясно, използвах функция parse, която преобразува стринг с Q израз в стойност, която може да бъде подадена на функция eval и която е необходима в функционалния селект. Също така отбелязваме, че preprocess е зададено като проекция (т.е. функция с частично определени аргументи) на функция select, един аргумент (таблица) е отсъстващ. Ако приложим preprocess към таблицата, ще получим компресирана таблица.

Вторият етап – е обновяване на агрегирана таблица. Нека напишем алгоритма първо в псевдокод:

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 вместо цикли е обичайно да се използват функции map/reduce. Но тъй като Q е векторен език и всички операции можем спокойно да прилагаме на всички символи наведнъж, в първоначалното приближение можем да се справим изобщо без цикли, извършвайки операции с всички символи едновременно:

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

За съжаление, за присвояване в таблица е необходим списък от редове, а не колони, и е необходимо да транспонираме матрицата (списък от колони в списък от редове) с помощта на функция flip. За голяма таблица това е скъпо, затова вместо това ще приложим обобщено присвояване на всяка колона отделно, използвайки функция map (която изглежда като апостроф):

.[aggTable;;:;]'[(idx;)each aggCols; (row[`high] | inputTable`high;row[`volume] + inputTable`volume;…)];

Отново използваме проекция на функция. Също така имайте предвид, че в Q създаването на списък е също функция и можем да я извикаме чрез функцията each(map), за да получим списък от списъци.

За да не бъде наборът от изчисляеми колони фиксиран, ще създадем израза динамично. Първо ще определим функциите за изчисление на всяка колона, използвайки променливите row и 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");

Някои колони са специални, първата им стойност не трябва да се изчислява чрез функция. Можем да определим, че тя е първа по колоната row[`numTrades] – ако в нея е 0, то стойността е първа. В Q има функция за избор — ?[Boolean list;list1;list2] – която избира стойност от списък 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 редът на изчисление е от дясно на ляво.

Сега имаме две основни функции, необходими за изчисленията; остава да добавим малко инфраструктура и услугата е готова.

Финални стъпки

Имаме функции preprocess и updateAgg, които вършат цялата работа. Но е необходимо също да осигурим правилен преход през минутите и да изчислим индексите за агрегация. Първо ще определим функцията 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, която ще променя текущата минута:

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 (традиционно име за тази функция за Q услуги), която се извиква от клиента за добавяне на данни:

upd:{[tblName;data] // tblName не е нужно, но обикновено услугата обработва няколко таблици 
  tm:exec distinct time from data:() xkey preprocess data; // preprocess & calc time
  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]; // обновим агрегирана таблица. Функцията ? търси индексите на елементите от списъка вдясно в списъка отляво.
 };

И ето, всичко е готово. Ето целия код на нашата услуга, както обещахме, само няколко реда:

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 – това са две минути. Може да опитате да стартирате първо за 1000 реда на всеки 25 милисекунди:

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 милиона реда в минута) е изключително голям, в такива случаи е обичайно да се стартират няколко клона (или дори десетки клонове) на сервиса, всеки от които обработва само част от символите. Въпреки това, резултатът е впечатляващ за интерпретируем език, който е ориентиран предимно към съхраняване на данни.

Може да се появи въпросът защо времето нараства нелинейно с размера на всяко обновление. Причината е, че компресионната функция всъщност е функция С, която работи много по-ефективно от updateAgg. От определен размер на обновлението (около 10,000) updateAgg достига своя таван и след това времето за изпълнение не зависи от размера на обновлението. Именно благодарение на предварителната стъпка Q услугата може да обработва такива обеми данни. Това подчертава колко е важно, когато работите с големи данни, да изберете правилния алгоритъм. Още един момент е правилното съхранение на данните в паметта. Ако данните не бяха организирани колонно или не бяха подредени във времето, бихме се сблъскали с нещо, наречено TLB cache miss – липса на адрес на страница в паметта в кеша на адресите на процесора. Търсенето на адреса отнема около 30 пъти повече време в случай на неуспех и при разпръснати данни може да забави услугата многократно.

Заключение

В тази статия показах, че базата данни KDB+ и Q са подходящи не само за съхранение на големи данни и лесен достъп до тях чрез селект, но и за създаване на услуги за обработка на данни, които могат да обработват стотици милиони редове / гигабайти данни дори в един единствен Q процес. Самият език Q позволява изключително кратко и ефективно реализиране на алгоритми, свързани с обработката на данни, благодарение на векторната си природа, вградения интерпретатор на диалекта SQL и много успешен набор от библиотечни функции.

Ще отбележа, че изложеното по-горе е само част от възможностите на Q, той има и други уникални характеристики. Например, изключително простият IPC протокол, който заличава границата между отделните Q процеси и позволява обединяване на стотици от тези процеси в единна мрежа, която може да се разположи на десетки сървъри в различни краища на света.

Източник: habr.com