Характеристики на езика Q и KDB+ на примера на услуга в реално време

Можете да прочетете какво представлява базата KDB+, езикът за програмиране Q, какви са техните силни и слаби страни в предишния ми Статия и накратко във въведението. В статията ще внедрим услуга на Q, която ще обработва входящия поток от данни и ще изчислява различни функции за агрегиране всяка минута в режим „в реално време“ (т.е. ще има време да изчисли всичко преди следващата порция данни). Основната характеристика на Q е, че той е векторен език, който ви позволява да работите не с единични обекти, а с техните масиви, масиви от масиви и други сложни обекти. Езици като Q и неговите роднини K, J, APL са известни със своята краткост. Често програма, която заема няколко екрана с код на познат език като Java, може да бъде написана на тях в няколко реда. Това е, което искам да демонстрирам в тази статия.

въведение

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

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

Проблем изявление

Има източник, който изпраща таблица с данни на всеки 25 милисекунди. Тъй като KDB+ се използва предимно във финансите, ще приемем, че това е таблица с транзакции (сделки), която има следните колони: време (време в милисекунди), сим (обозначението на компанията на борсата - IBM, AAPL,…), цена (цената, на която са закупени акциите), размер (размер на сделката). Интервалът от 25 милисекунди е произволен, не е твърде малък и не е твърде дълъг. Неговото присъствие означава, че данните идват в услугата вече буферирани. Би било лесно да се приложи буфериране от страна на услугата, включително динамично буфериране в зависимост от текущото натоварване, но за простота ще се съсредоточим върху фиксиран интервал.

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

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

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

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

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

Нека незабавно да обсъдим една неочевидна точка - как да инициализираме тези колони за първи път и за всяка следваща минута. Някои колони от типа firstPrice трябва да се инициализират до нула всеки път; тяхната стойност е недефинирана. Другите типове сила на звука винаги трябва да бъдат зададени на 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 (отдясно наляво). Целта е да се гарантира, че изчислението преминава от висок към cumVolume, тъй като някои колони зависят от предишните.

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

rollColumns:`sym`cumVolume;

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

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

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

accumulatorCols:`numTrades`volume`pvolume`turnover;
specialCols:`high`low`firstPrice`firstSize;

Ред на изчисление

Ще актуализираме обобщената таблица на два етапа. За ефективност първо свиваме входящата таблица, така че да има само един ред за всеки знак и минута. Фактът, че всички наши функции са инкрементални и асоциативни, гарантира, че резултатът от тази допълнителна стъпка няма да се промени. Можете да намалите таблицата, като използвате select:

select high:max price, low:min price … by sym,time.minute from table

Този метод има недостатък - наборът от изчислени колони е предварително дефиниран. За щастие, в Q, select също е имплементиран като функция, където можете да замествате динамично създадени аргументи:

?[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 и която е необходима във функцията select. Също така имайте предвид, че предварителният процес се дефинира като проекция (т.е. функция с частично дефинирани аргументи) на функцията за избор, един аргумент (таблицата) липсва. Ако приложим предварителна обработка към таблица, ще получим компресирана таблица.

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

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

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

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

Последни стъпки

Имаме функции за предварителна обработка и 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[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 XNUMX броя и да добавим функция за създаване на произволна таблица. Във втората конзола:

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 XNUMX:

start 10000

Резултати:

min| 00:00:00.004
avg| 9.191458
med| 9f
max| 00:00:00.030

Отново нищо особено, но това са 24 милиона реда в минута, 400 хиляди в секунда. За повече от 25 милисекунди актуализацията се забави само 5 пъти, очевидно при смяна на минутата. Нека увеличим до 100.000 XNUMX:

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

Може да възникне въпросът защо времето расте нелинейно с размера на всяка актуализация. Причината е, че функцията за свиване всъщност е C функция, която е много по-ефективна от updateAgg. Започвайки от определен размер на актуализацията (около 10.000 30), updateAgg достига своя таван и след това времето за изпълнение не зависи от размера на актуализацията. Това се дължи на предварителната стъпка Q, че услугата е в състояние да усвои такива обеми от данни. Това подчертава колко е важно да изберете правилния алгоритъм при работа с големи данни. Друг момент е правилното съхранение на данни в паметта. Ако данните не се съхраняват колонно или не се подреждат по време, тогава щяхме да се запознаем с такова нещо като пропуск на TLB кеш - липсата на адрес на страница от паметта в адресния кеш на процесора. Търсенето на адрес отнема около XNUMX пъти повече време, ако е неуспешно, а ако данните са разпръснати, това може да забави услугата няколко пъти.

Заключение

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

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

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