Възможностите на езика 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,…), цена (price, по която са закупени акциите), размер (size на сделката). Интервалът от 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 ("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];

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

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

for each sym in inputTable
  idx: редов индекс в агрегираната таблица за 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]; // създавame празна типизирана таблица, 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; // предварителна обработка & изчисляване на времето
  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]; // обновяваме агрегирана таблица. Функцията ? търси индекс на елементите от списъка вдясно в списъка вляво.
 };

Вот и все. Вот полный код нашего сервиса, как и обещалось, всего несколько строк:

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