You can read about what KDB+ is, the Q programming language, and their strengths and weaknesses in my previous and briefly in the introduction. In this article, we will implement a service in Q that will process an incoming stream of data and calculate various aggregate functions in a 'real-time' mode (i.e., it will manage to compute everything before the next batch of data arrives). The main feature of Q is that it is a vector language, allowing manipulation of not just individual objects, but arrays of these objects, arrays of arrays, and other complex structures. Languages like Q and its relatives K, J, and APL are known for their brevity. Often, a program that spans several screens of code in a familiar language like Java can be expressed in just a few lines with them. That is exactly what I want to demonstrate in this article.
Introduction
KDB+ is a column-oriented database designed for very large volumes of data organized in a specific way (primarily by time). It is mainly used in financial institutions – banks, investment funds, and insurance companies. The Q language is the internal language of KDB+ that allows for efficient work with this data. The ideology of Q emphasizes brevity and efficiency, often sacrificing clarity. This is justified by the fact that a vector language is inherently complex to grasp, and the brevity and richness of the syntax allow for a much larger portion of the program to be seen on a single screen, ultimately aiding understanding.
In this article, we will implement a full program in Q, and you may want to try it out yourself. For this, you will need Q itself. You can download the free 32-bit version from the kx company website – . There, if you are interested, you will also find reference information about Q, the book , and various articles on this topic.
Task Definition
There is a source that sends a table with data every 25 milliseconds. Since KDB+ is primarily used in finance, let’s assume this is a trades table, which has the following columns: time (milliseconds), sym (stock symbol – IBM, AAPL,…), price (the price at which the shares were purchased), size (the size of the transaction). The 25-millisecond interval is chosen arbitrarily; it is neither too short nor too long. Its existence indicates that the data is already buffered when arriving at the service. It would be easy to implement buffering on the service side, including dynamic buffering depending on the current load, but for simplicity, we will stick to a fixed interval.
The service must compute per minute for each incoming symbol from the sym column a set of aggregate functions – max price, avg price, sum size, and other useful information. For simplicity, we will assume that all functions can be computed incrementally, that is, to obtain a new value, it is enough to know two numbers – the old and the incoming value. For example, the max, average, and sum functions have this property, while the median function does not.
We will also assume that the incoming data stream is ordered by time. This allows us to work only with the last minute. In practice, it is sufficient to be able to operate with the current and the previous minutes in case some updates are delayed. For simplicity, we will not consider this case.
Aggregate Functions
Below are the necessary aggregate functions. I have included as many as possible to increase the load on the service:
- high – max price – the maximum price per minute.
- low – min price – the minimum price per minute.
- firstPrice – first price – the first price per minute.
- lastPrice – last price – the last price per minute.
- firstSize – first size – the first transaction size per minute.
- lastSize – last size – the last transaction size per minute.
- numTrades – count i – the number of transactions per minute.
- volume – sum size – the total transaction size per minute.
- pvolume – sum price – the total price per minute, necessary for avgPrice.
- turnover – sum price*size – the total transaction volume per minute.
- avgPrice – pvolume%numTrades – the average price per minute.
- avgSize – volume%numTrades – the average transaction size per minute.
- vwap – turnover%volume – the volume-weighted average price per minute.
- cumVolume – sum volume – the accumulated transaction size over all time.
Let's immediately discuss a non-obvious point – how to initialize these columns for the first time and for each subsequent minute. Some columns like firstPrice need to be initialized with a null value every time, as their value is undefined. Others like volume should always be set to 0. There are also columns that require a combined approach – for example, cumVolume needs to be copied from the previous minute, while for the first minute, it should be set to 0. We will define all these parameters using the dictionary data type (analogous to a record):
// 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 объяснен ниже
I have added sym and time to the dictionary for convenience, now initWith is a complete line from the final aggregated table, where we just need to set the correct sym and time. It can be used to add new rows to the table.
aggCols will be needed when creating the aggregation function. The list needs to be inverted due to the peculiarities of expression evaluation order in Q (right to left). The goal is to ensure computation flows from high to cumVolume, as some columns depend on the previous ones.
Columns that need to be copied to the new minute from the previous one, the sym column has been added for convenience:
rollColumns:`sym`cumVolume;
Now let's divide the columns into groups according to how they should be updated. We can distinguish three types:
- Accumulators (volume, turnover, ...) – we need to add the incoming value to the previous one.
- Special points (high, low, ...) – the first value in the minute is taken from incoming data, the others are calculated using a function.
- Others. Always calculated using a function.
Let's define variables for these classes:
accumulatorCols:`numTrades`volume`pvolume`turnover;
specialCols:`high`low`firstPrice`firstSize;
Order of calculations
We will update the aggregated table in two stages. For efficiency, we will first compress the incoming table so that there is one row left for each symbol and minute. The fact that all our functions are incremental and associative guarantees us that the result from this additional step will not change. The table could be compressed using a select:
select high:max price, low:min price … by sym,time.minute from table
The downside of this method is that the set of computed columns is predetermined. Fortunately, in Q the select is also implemented as a function, which can accept dynamically created arguments:
?[table;whereClause;byClause;selectClause]
I won't describe the argument format in detail; in our case, the only non-trivial aspects will be the by and select expressions, and they should be dictionaries of the form columns!expressions. Thus, a compressing function can be defined as follows:
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 is the map function in Q for a single list
preprocess:?[;();`sym`time!`sym`time.minute;selExpression];
For clarity, I used the parse function, which transforms a string with a Q expression into a value that can be passed to the eval function and required in a functional select. We should also note that preprocess is defined as a projection (i.e., a function with partially defined arguments) of the select function, with one argument (table) missing. If we apply preprocess to the table, we will get a compressed table.
The second stage is updating the aggregated table. Let's first write the algorithm in pseudocode:
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];
…
In Q, it is customary to use map/reduce functions instead of loops. However, since Q is a vector language and we can comfortably apply all operations to all symbols at once, we can initially do without a loop, performing operations on all symbols simultaneously:
idx:calcIdx inputTable;
row:aggTable idx;
aggTable[idx;`high]: row[`high] | inputTable`high;
aggTable[idx;`volume]: row[`volume] + inputTable`volume;
…
But we can go further; Q has a unique and exceptionally powerful operator—the generalized assignment operator. It allows changing a set of values in a complex data structure using a list of indices, functions, and arguments. In our case, it looks like this:
idx:calcIdx inputTable;
rows:aggTable idx;
// .[target;(idx0;idx1;..);function;argument] ~ target[idx 0;idx 1;…]: function[target[idx 0;idx 1;…];argument], in our case, the function is assignment
.[aggTable;(idx;aggCols);:;flip (row[`high] | inputTable`high;row[`volume] + inputTable`volume;…)];
Unfortunately, for assignment in the table, a list of rows is needed instead of columns, necessitating transposing the matrix (converting a list of columns to a list of rows) using the flip function. For a large table, this is costly, so instead, we will apply generalized assignment to each column separately, using the map function (which looks like an apostrophe):
[aggTable;;:;]'[(idx;)each aggCols; (row[`high] | inputTable`high;row[`volume] + inputTable`volume;…)]
We are using the function projection again. Note that in Q, creating a list is also a function, and we can invoke it using the each(map) function to obtain a list of lists.
To ensure that the set of calculated columns is not fixed, we'll dynamically create the expression above. First, we'll define functions for calculating each column, using the row and inp variables to reference the aggregated and input data:
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");
Some columns are special; their first value should not be calculated by the function. We can determine that it is the first by checking the row[`numTrades] column – if it is 0, then the value is the first. In Q, there is a selection function — ?[Boolean list;list1;list2] — which selects a value from list 1 or 2 depending on the condition in the first argument:
// high -> ?[isFirst;inp`high;row[`high]|inp`high]
// @ - тоже обобщенное присваивание для случая когда индекс неглубокий
@[`aggExpression;specialCols;{[x;y]"?[isFirst;inp`",y,";",x,"]"};string specialCols];
Here I invoked generalized assignment with my function (an expression in curly braces). It takes the current value (the first argument) and an additional argument that I pass in the 4th parameter.
Separately, let's add the accumulator columns since they share the same function:
// volume -> row[`volume]+inp`volume
aggExpression[accumulatorCols]:{"row[`",x,"]+inp`",x } each string accumulatorCols;
This is a common assignment by Q standards, except I'm assigning a list of values all at once. Finally, let's create the main function:
// ":",/: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),")]}";
This expression dynamically creates a function from a string that contains the expression I mentioned above. The result will look like this:
{[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])]}
The order of column calculations is inverted, as in Q the order of calculations goes from right to left.
Now we have the two main functions necessary for calculations, and all that's left is to add a bit of infrastructure, and the service will be ready.
Final steps
We have preprocess and updateAgg functions that do all the work. However, we need to ensure proper transitions through the minutes and calculate the indices for aggregation. First, let's define the init function:
init:{
tradeAgg:: 0#enlist[initWith]; // creating an empty typed table, enlist converts the dictionary to a table, and 0# means to take 0 elements from it
currTime::00:00; // starting at 0, :: indicates assignment to a global variable
currSyms::`u#`symbol$(); // `u# - converts the list into a tree for faster element searching
offset::0; // index in tradeAgg where the current minute starts
rollCache:: `sym xkey update `u#sym from rollColumns#tradeAgg; // cache for the latest values of roll columns, table with key sym
}
We will also define the roll function, which will change the current minute:
roll:{[tm]
if[currTime>tm; :init[]]; // if we've crossed midnight, just call init
rollCache,::offset _ rollColumns#tradeAgg; // update the cache – take roll columns from aggTable, trim, insert into rollCache
offset::count tradeAgg;
currSyms::`u#`$();
}
We will need a function to add new symbols:
addSyms:{[syms]
currSyms,::syms; // add to the list of known ones
// add to the table sym, time, and rollColumns using generalized assignment.
// The function ^ substitutes default values for roll columns if the symbol is not in the cache. value flip table returns a list of columns in the table.
`tradeAgg upsert @[count[syms]#enlist initWith;`sym`time,cols rc;:;(syms;currTime), (initWith cols rc)^value flip rc:rollCache ([] sym: syms)];
}
And finally, the upd function (a traditional name for this function in Q services), which is called by the client to add data:
upd:{[tblName;data] // tblName is not needed, but usually the service processes multiple tables
tm:exec distinct time from data:() xkey preprocess data; // preprocess & calc time
updMinute[data] each tm; // add data for each minute
};
updMinute:{[data;tm]
if[tmcurrTime; roll tm; currTime::tm]; // change the minute if necessary
data:select from data where time=tm; // filtering
if[count msyms:syms where not (syms:data`sym)in currSyms; addSyms msyms]; // new symbols
updateAgg[`tradeAgg;offset+currSyms?syms;data]; // update the aggregated table. The function ? finds the index of the elements in the list on the right in the list on the left.
};
That's it. Here is the complete code of our service, as promised, just a few lines:
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];
};
Testing
Let's check the service's performance. To do this, we will run it in a separate process (place the code in the file service.q) and call the init function:
q service.q –p 5566
q)init[]
In another console, run a second Q process and connect to the first one:
h:hopen `:host:5566
h:hopen 5566 // if both are on the same host
First, let's create a list of symbols – 10,000 of them – and add a function to generate a random table. In the second console:
syms:`IBM`AAPL`GOOG,-9997?`8
rnd:{[n;t] ([] sym:n?syms; time:t+asc n#til 25; price:n?10f; size:n?10)}
I added three real symbols to the list to make it easier to find them in the table. The rnd function generates a random table with n rows, where time varies from t to t+25 milliseconds.
Now we can try to send data to the service (let's add the first ten hours):
{h (`upd;`trade;rnd[10000;x])} each `time$00:00 + til 60*10
You can check in the service that the table has been updated:
c 25 200
select from tradeAgg where sym=`AAPL
-20#select from tradeAgg where sym=`AAPL
Result:
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|2919Now let's conduct load testing to find out how much data the service can handle per minute. I remind you that we set the update interval to 25 milliseconds. Accordingly, the service should (on average) take no more than 20 milliseconds per update to give users time to request data. Enter the following in the second process:
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 is two minutes. You can try running it first for 1000 rows every 25 milliseconds:
start 1000
In my case, the result is around a couple of milliseconds per update. So I will immediately increase the number of rows to 10,000:
start 10000
Result:
min| 00:00:00.004
avg| 9.191458
med| 9f
max| 00:00:00.030
Again, nothing special, yet this is 24 million rows per minute, 400 thousand per second. More than 25 milliseconds for the update only caused delays 5 times, apparently when switching minutes. Let's increase to 100,000:
start 100000
Result:
min| 00:00:00.013
avg| 25.11083
med| 24f
max| 00:00:00.108
q)sum times
00:02:00.532
As we can see, the service is barely keeping up, but it manages to stay afloat nonetheless. Such a volume of data (240 million rows per minute) is extraordinarily large; in such cases, it is common to run several clones (or even dozens of clones) of the service, each processing only a part of the symbols. Nevertheless, the result is impressive for an interpreted language primarily focused on data storage.
One might wonder why the time grows non-linearly with the size of each update. The reason is that the compressing function is actually a C function that works much more efficiently than updateAgg. Starting from a certain update size (around 10,000), updateAgg reaches its ceiling, and further, its execution time does not depend on the update size. It is precisely due to the preliminary Q step that the service can process such volumes of data. This highlights the importance of choosing the right algorithm when working with big data. Another point is the correct storage of data in memory. If the data were not stored column-wise or not ordered by time, we would encounter something known as TLB cache miss — the absence of a memory page address in the CPU's address cache. In case of a miss, finding the address takes about 30 times longer, and in the case of scattered data, it can slow down the service several times over.
Conclusion
In this article, I demonstrated that the KDB+ database and Q are suitable not only for storing large data sets and simple access through selects but also for creating data processing services capable of handling hundreds of millions of rows/gigabytes of data even within a single Q process. The Q language allows for remarkably brief and effective implementation of data processing algorithms due to its vector nature, built-in SQL dialect interpreter, and an excellent set of library functions.
I would like to note that what has been presented above is just a fraction of Q's capabilities; it has other unique features as well. For instance, an incredibly simple IPC protocol that erases the boundary between separate Q processes and allows hundreds of these processes to be combined into a single network that can be distributed across dozens of servers around the globe.
Source: habr.com
