Карактеристике К и КДБ+ језика на примеру услуге у реалном времену

О томе шта је КДБ+ база, К програмски језик, које су њихове предности и мане можете прочитати у мом претходном Чланак и укратко у уводу. У чланку ћемо имплементирати услугу на К која ће обрадити долазни ток података и израчунати различите функције агрегације сваког минута у режиму „реалног времена“ (тј. имаће времена да све израчуна пре следећег дела података). Главна карактеристика К је да је то векторски језик који вам омогућава да радите не са појединачним објектима, већ са њиховим низовима, низовима низова и другим сложеним објектима. Језици као што су К и његови сродници К, Ј, АПЛ познати су по својој краткоћи. Често се програм који заузима неколико екрана кода на познатом језику као што је Јава може написати на њима у неколико редова. То је оно што желим да покажем у овом чланку.

Увод

КДБ+ је колонарна база података фокусирана на веома велике количине података, пореданих на специфичан начин (пре свега по времену). Користи се првенствено у финансијским институцијама - банкама, инвестиционим фондовима, осигуравајућим друштвима. К језик је интерни језик КДБ+ који вам омогућава да ефикасно радите са овим подацима. К идеологија је краткоћа и ефикасност, док се јасноћа жртвује. То се оправдава чињеницом да ће векторски језик у сваком случају бити тешко разумљив, а сажетост и богатство снимка омогућава вам да на једном екрану видите много већи део програма, што га на крају чини лакшим за разумевање.

У овом чланку имплементирамо пуноправни програм у К и можда бисте желели да га испробате. Да бисте то урадили, биће вам потребан прави К. Можете преузети бесплатну 32-битну верзију на веб локацији компаније кк – ввв.кк.цом. Тамо, ако сте заинтересовани, наћи ћете референтне информације о књизи К К За смртнике и разне чланке на ову тему.

Проблем статемент

Постоји извор који шаље табелу са подацима сваких 25 милисекунди. Пошто се КДБ+ користи првенствено у финансијама, претпоставићемо да се ради о табели трансакција (тргова), која има следеће колоне: време (време у милисекундама), сим (ознака компаније на берзи - ИБМ-, ААПЛ,…), цена (цена по којој су акције купљене), величина (величина трансакције). Интервал од 25 милисекунди је произвољан, није премали и не предугачак. Његово присуство значи да подаци долазе у сервис већ баферовани. Било би лако имплементирати баферовање на страни услуге, укључујући динамичко баферовање у зависности од тренутног оптерећења, али ради једноставности, фокусираћемо се на фиксни интервал.

Услуга мора да броји сваки минут за сваки долазни симбол из колоне сим скуп функција агрегирања - максимална цена, просечна цена, величина суме, итд. корисна информација. Ради једноставности, претпоставићемо да се све функције могу израчунати инкрементално, тј. да бисте добили нову вредност, довољно је знати два броја - стари и долазне вредности. На пример, функције мак, просек, збир имају ово својство, али функција медијана не.

Такође ћемо претпоставити да је долазни ток података временски распоређен. Ово ће нам дати прилику да радимо само у последњем тренутку. У пракси је довољно да можете да радите са тренутним и претходним минутама у случају да нека ажурирања касне. Ради једноставности, нећемо разматрати овај случај.

Функције агрегације

Потребне функције агрегације су наведене у наставку. Узео сам што је више могуће да повећам оптерећење услуге:

• висока – максимална цена – максимална цена по минуту.
• ниска – мин цена – минимална цена по минуту.
• фирстПрице – прва цена – прва цена по минуту.
• ластПрице – последња цена – последња цена по минуту.
• фирстСизе – прва величина – прва величина трговине у минути.
• ластСизе – последња величина — последња величина трговине за минут.
• нумТрадес – цоунт и – број трговина у минути.
• волуме – сума сизе – збир величина трговине у минути.
• пволуме – збир цена – збир цена по минуту, потребан за авгПрице.
• – збирна цена промета*величина – укупан обим трансакција по минуту.
• авгПрице – пволуме%нумТрадес – просечна цена по минуту.
• авгСизе – волуме%нумТрадес – просечна величина трговине у минути.
• ввап – обрт%волуме – просечна цена по минуту пондерисана величином трансакције.
• цумВолуме – збирни обим – акумулирана величина трансакција током читавог времена.

Хајде одмах да разговарамо о једној неочигледној тачки - како да иницијализујемо ове колоне први пут и за сваки следећи минут. Неке колоне типа фирстПрице морају бити иницијализоване на нулл сваки пут; њихова вредност је недефинисана. Други типови волумена увек морају бити подешени на 0. Постоје и колоне које захтевају комбиновани приступ – на пример, цумВолуме се мора копирати из претходног минута, а за први постављен на 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 объяснен ниже

Додао сам сим и време у речник ради погодности, сада је инитВитх готова линија из коначне агрегиране табеле, где остаје да се подеси тачан сим и време. Можете га користити за додавање нових редова у табелу.

Биће нам потребни аггЦолс када креирамо функцију агрегације. Листа мора бити обрнута због редоследа којим се процењују изрази у К (с десна на лево). Циљ је да се обезбеди да прорачун иде од високог до цумВолуме, пошто неке колоне зависе од претходних.

Колоне које треба копирати у нови минут од претходног, колона сим се додаје ради погодности:

rollColumns:`sym`cumVolume;

Хајде сада да поделимо колоне у групе према томе како треба да се ажурирају. Могу се разликовати три типа:

1. Акумулатори (запремина, промет,..) – морамо додати улазну вредност претходној.
2. Са посебном тачком (висока, ниска, ..) – прва вредност у минути се узима из долазних података, остатак се израчунава помоћу функције.
3. Одмори се. Увек се израчунава помоћу функције.

Хајде да дефинишемо променљиве за ове класе:

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

Редослед израчунавања

Агрегирану табелу ћемо ажурирати у две фазе. Ради ефикасности, прво смањујемо долазну табелу тако да постоји само један ред за сваки знак и минут. Чињеница да су све наше функције инкременталне и асоцијативне гарантује да се резултат овог додатног корака неће променити. Можете да смањите табелу користећи селецт:

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

Овај метод има недостатак - скуп израчунатих колона је унапред дефинисан. На срећу, у К-у, селецт је такође имплементиран као функција у којој можете заменити динамички креиране аргументе:

?[table;whereClause;byClause;selectClause]

Нећу детаљно описивати формат аргумената, у нашем случају, само би и селецт изрази ће бити нетривијални и то треба да буду речници облика колоне!изрази. Дакле, функција скупљања се може дефинисати на следећи начин:

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

Ради јасноће, користио сам функцију парсе, која претвара стринг са К изразом у вредност која се може пренети функцији евал и која је потребна у функцији селецт. Такође имајте на уму да је предпроцес дефинисан као пројекција (тј. функција са делимично дефинисаним аргументима) функције за избор, један аргумент (табела) недостаје. Ако применимо претпроцес на табелу, добићемо компресовану табелу.

Друга фаза је ажурирање збирне табеле. Хајде да прво напишемо алгоритам у псеудокоду:

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

У К, уобичајено је користити функције мапирања/редуцирања уместо петљи. Али пошто је К векторски језик и можемо лако да применимо све операције на све симболе одједном, онда у првој апроксимацији можемо уопште без петље, извршавајући операције на свим симболима одједном:

idx:calcIdx inputTable;
row:aggTable idx;
aggTable[idx;`high]: row[`high] | inputTable`high;
aggTable[idx;`volume]: row[`volume] + inputTable`volume;
…

Али можемо ићи даље, К има јединствен и изузетно моћан оператор - генерализовани оператор доделе. Омогућава вам да промените скуп вредности у сложеној структури података користећи листу индекса, функција и аргумената. У нашем случају то изгледа овако:

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

Нажалост, да бисте доделили табели, потребна вам је листа редова, а не колона, и морате да транспонујете матрицу (листу колона у листу редова) користећи функцију флип. Ово је скупо за велику табелу, па уместо тога примењујемо генерализовани задатак на сваку колону посебно, користећи функцију мапе (која изгледа као апостроф):

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

Поново користимо пројекцију функције. Такође имајте на уму да је у К креирање листе такође функција и можемо је позвати помоћу функције евери(мап) да бисмо добили листу листа.

Да бисмо осигурали да скуп израчунатих колона није фиксан, креираћемо горњи израз динамички. Хајде да прво дефинишемо функције за израчунавање сваке колоне, користећи променљиве реда и инп за упућивање на агрегиране и улазне податке:

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");

Неке колоне су посебне; њихова прва вредност не би требало да се израчунава помоћу функције. Можемо утврдити да је први по колони ров[`нумТрадес] - ако садржи 0, онда је вредност прва. К има функцију избора - ?[Боолеан лист;лист1;лист2] - која бира вредност са листе 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;

Ово је нормалан задатак по К стандардима, али ја додељујем листу вредности одједном. На крају, направимо главну функцију:

// ":",/: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])]}

Редослед вредновања колоне је обрнут јер је у К редослед евалуације с десна на лево.

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

Завршни кораци

Имамо препроцес и упдатеАгг функције које обављају сав посао. Али и даље је неопходно обезбедити исправан прелаз кроз минуте и израчунати индексе за агрегацију. Пре свега, хајде да дефинишемо инит функцију:

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:{[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:{[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];
 };

Тестирање

Хајде да проверимо перформансе услуге. Да бисмо то урадили, покренимо га у посебном процесу (ставите код у датотеку сервице.к) и позовимо функцију инит:

q service.q –p 5566

q)init[]

У другој конзоли покрените други К процес и повежите се са првим:

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)}

Додао сам три права симбола на листу како бих их лакше тражио у табели. Функција рнд креира случајну табелу са н редова, где време варира од т до т+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 милиона редова у минути) је изузетно велики, у таквим случајевима уобичајено је да се покрене неколико клонова (или чак десетине клонова) сервиса, од којих сваки обрађује само део карактера. Ипак, резултат је импресиван за интерпретирани језик који се првенствено фокусира на складиштење података.

Може се поставити питање зашто време расте нелинеарно са величином сваког ажурирања. Разлог је тај што је функција скупљања заправо Ц функција, која је много ефикаснија од упдатеАгг-а. Почевши од одређене величине ажурирања (око 10.000), упдатеАгг достиже свој плафон и тада његово време извршења не зависи од величине ажурирања. Због прелиминарног корака К услуга је у стању да свари такве количине података. Ово наглашава колико је важно одабрати прави алгоритам када радите са великим подацима. Друга ствар је исправно складиштење података у меморији. Ако подаци нису похрањени у колонама или нису поређани по времену, онда бисмо се упознали са таквом ствари као што је промашај ТЛБ кеша - одсуство адресе меморијске странице у кешу адреса процесора. Тражење адресе траје око 30 пута дуже ако је неуспешно, а ако су подаци разбацани, може неколико пута успорити услугу.

Закључак

У овом чланку сам показао да су КДБ+ и К база података погодне не само за складиштење великих података и лак приступ њима путем селекције, већ и за креирање услуга обраде података које су способне да обрађују стотине милиона редова/гигабајта података чак иу један једини К процес. Сам К језик омогућава изузетно концизну и ефикасну имплементацију алгоритама везаних за обраду података због своје векторске природе, уграђеног тумача СКЛ дијалекта и веома успешног скупа библиотечких функција.

Приметићу да је горе наведено само део онога што К може да уради, има и друге јединствене карактеристике. На пример, изузетно једноставан ИПЦ протокол који брише границу између појединачних К процеса и омогућава вам да комбинујете стотине ових процеса у једну мрежу, која се може налазити на десетинама сервера у различитим деловима света.

Извор: ввв.хабр.цом