Sobre qué es la base KDB+, el lenguaje de programación Q, cuáles son sus fortalezas y debilidades, se puede leer en mi anterior y brevemente en la introducción. En este artículo, implementaremos en Q un servicio que procesará un flujo de datos entrante y calculará diversas funciones agregadas minuto a minuto en tiempo real (es decir, se encargará de calcular todo antes de recibir el siguiente lote de datos). La principal característica de Q es que es un lenguaje vectorial que permite operar no solo con objetos individuales, sino también con sus matrices, matrices de matrices y otros objetos compuestos complejos. Lenguajes como Q, junto con sus parientes K, J, APL, son famosos por su brevedad. A menudo, un programa que ocuparía varias pantallas de código en lenguajes comunes como Java puede escribirse en ellos en unas pocas líneas. Precisamente esto es lo que quiero demostrar en este artículo.
Introducción
KDB+ es una base de datos en columna orientada a volumenes de datos muy grandes, organizados de cierta manera (principalmente por tiempo). Se utiliza, en primer lugar, en organizaciones financieras: bancos, fondos de inversión, compañías de seguros. El lenguaje Q es el lenguaje interno de KDB+ que permite trabajar eficazmente con estos datos. La ideología de Q es la brevedad y la eficiencia, sacrificando la claridad. Esto se justifica porque un lenguaje vectorial será complicado de apreciar de todos modos, y la brevedad y la riqueza de la notación permiten ver en una única pantalla una parte mucho mayor del programa, facilitando su comprensión.
En este artículo implementaremos un programa completo en Q y, posiblemente, querrás probarlo en acción. Para ello, necesitarás Q. Puedes descargar la versión gratuita de 32 bits en el sitio web de la empresa kx - . Allí también, si te interesa, encontrarás información de referencia sobre Q, el libro y diversos artículos al respecto.
Planteamiento del problema
Hay una fuente que envía cada 25 milisegundos una tabla con datos. Dado que KDB+ se aplica principalmente en finanzas, supongamos que esta es una tabla de transacciones (trades), que contiene las siguientes columnas: time (tiempo en milisegundos), sym (símbolo de la empresa en la bolsa - IBM, AAPL,…), price (precio, al que se compraron las acciones), size (tamaño de la transacción). El intervalo de 25 milisegundos se elige de manera arbitraria, no es demasiado pequeño ni demasiado grande. Su existencia significa que los datos llegan al servicio ya en búfer. Sería fácil implementar el almacenamiento en búfer en el lado del servicio, incluida la dinámica, dependiendo de la carga actual, pero por simplicidad nos quedaremos con un intervalo fijo.
El servicio debe calcular por minuto para cada símbolo que llegue de la columna sym un conjunto de funciones de agregación: max price, avg price, sum size, etc. información útil. Por simplicidad, supongamos que todas las funciones se pueden calcular de manera incremental, es decir, para obtener un nuevo valor solo es necesario conocer dos números: el valor antiguo y el valor entrante. Por ejemplo, las funciones max, average, sum tienen esta propiedad, mientras que la función de mediana no.
También supongamos que el flujo de datos entrante está ordenado por tiempo. Esto nos permitirá trabajar solo con el último minuto. En la práctica, es suficiente saber trabajar con el minuto actual y el anterior, en caso de que haya algunos retrasos en las actualizaciones. Por simplicidad, no consideraremos este caso.
Funciones de agregación
A continuación se enumeran las funciones de agregación necesarias. He tomado tantas como sea posible para aumentar la carga en el servicio:
- high – max price – precio máximo por minuto.
- low – min price – precio mínimo por minuto.
- firstPrice – first price – primer precio por minuto.
- lastPrice – last price – último precio por minuto.
- firstSize – first size – primer tamaño de la transacción por minuto.
- lastSize – last size – último tamaño de la transacción por minuto.
- numTrades – count i – número de transacciones por minuto.
- volume – sum size – suma de los tamaños de las transacciones por minuto.
- pvolume – sum price – suma de precios por minuto, necesaria para avgPrice.
- turnover – sum price*size – volumen total de transacciones por minuto.
- avgPrice – pvolume%numTrades – precio promedio por minuto.
- avgSize – volume%numTrades – tamaño promedio de la transacción por minuto.
- vwap – turnover%volume – precio promedio ponderado por tamaño de transacción por minuto.
- cumVolume – sum volume – tamaño acumulado de transacciones desde el principio.
Hablemos de un punto no tan obvio: cómo inicializar estas columnas por primera vez y para cada minuto siguiente. Algunas columnas, como firstPrice, deben inicializarse cada vez con un valor nulo, ya que su valor no está definido. Otras, como volume, deben establecerse siempre en 0. También hay columnas que requieren un enfoque combinado: por ejemplo, cumVolume debe copiarse del minuto anterior, mientras que para el primero se debe establecer en 0. Estableceremos todos estos parámetros utilizando un tipo de datos diccionario (análogo a un registro):
// 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 объяснен ниже
He añadido sym y time al diccionario para mayor comodidad; ahora initWith es una línea lista de la tabla agregada final, donde solo queda especificar los correctos sym y time. Se puede utilizar para agregar nuevas filas a la tabla.
aggCols los necesitaremos al crear la función de agregación. La lista debe invertirse debido a las particularidades del orden de evaluación de las expresiones en Q (de derecha a izquierda). El objetivo es asegurar el cálculo en la dirección de high a cumVolume, ya que algunas columnas dependen de las anteriores.
Las columnas que deben copiarse al nuevo minuto del anterior, se ha añadido la columna sym para mayor comodidad:
rollColumns:`sym`cumVolume;
Ahora dividiremos las columnas en grupos según cómo deben actualizarse. Se pueden identificar tres tipos:
- Acumuladores (volume, turnover,..) - debemos sumar el valor entrante al anterior.
- Con un punto especial (high, low, ..) - el primer valor del minuto se toma de los datos entrantes, los demás se calculan a través de la función.
- Los demás. Siempre se calculan mediante la función.
Definamos las variables para estas clases:
accumulatorCols:`numTrades`volume`pvolume`turnover;
specialCols:`high`low`firstPrice`firstSize;
Orden de cálculos
Actualizaremos la tabla agregada en dos etapas. Para mayor eficiencia, primero compactaremos la tabla de entrada de modo que quede una sola fila para cada símbolo y minuto. El hecho de que todas nuestras funciones sean incrementales y asociativas nos garantiza que el resultado de este paso adicional no cambiará. La tabla se podría compactar mediante un select:
select high:max price, low:min price … by sym,time.minute from table
Este método tiene una desventaja: el conjunto de columnas calculables está definido de antemano. Afortunadamente, en Q, el select se implementa también como una función, donde se pueden introducir argumentos creados dinámicamente:
?[table;whereClause;byClause;selectClause]
No voy a describir detalladamente el formato de los argumentos, en nuestro caso, solo las expresiones by y select serán no triviales y deben ser diccionarios del tipo columns!expressions. Así, se puede definir la función de compresión de la siguiente manera:
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 es una función map en Q para una lista
preprocess:?[;();`sym`time!`sym`time.minute;selExpression];
Para mayor claridad, utilicé la función parse, que convierte una cadena con una expresión Q en un valor que puede ser pasado a la función eval y que se requiere en el select funcional. También notemos que preprocess está definido como una proyección (es decir, una función con argumentos parcialmente definidos) de la función select, ya que falta un argumento (la tabla). Si aplicamos preprocess a la tabla, obtendremos una tabla comprimida.
La segunda etapa es la actualización de la tabla agregada. Primero escribiremos el algoritmo en pseudocódigo:
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];
…
En Q, es común utilizar funciones map/reduce en lugar de bucles. Pero dado que Q es un lenguaje vectorial, podemos aplicar fácilmente todas las operaciones a todos los símbolos a la vez, por lo que, en un primer acercamiento, podemos prescindir completamente del bucle, realizando operaciones sobre todos los símbolos simultáneamente:
idx:calcIdx inputTable;
row:aggTable idx;
aggTable[idx;`high]: row[`high] | inputTable`high;
aggTable[idx;`volume]: row[`volume] + inputTable`volume;
…
Pero podemos avanzar aún más, en Q hay un operador único y extremadamente poderoso: el operador de asignación generalizada. Permite modificar un conjunto de valores en una estructura de datos compleja utilizando una lista de índices, funciones y argumentos. En nuestro caso, se ve así:
idx:calcIdx inputTable;
rows:aggTable idx;
// .[target;(idx0;idx1;..);function;argument] ~ target[idx 0;idx 1;…]: function[target[idx 0;idx 1;…];argument], en nuestro caso la función es la asignación
.[aggTable;(idx;aggCols);:;flip (row[`high] | inputTable`high;row[`volume] + inputTable`volume;…);
Desafortunadamente, para asignar a la tabla se necesita una lista de filas, no de columnas, y es necesario transponer la matriz (convertir una lista de columnas en una lista de filas) utilizando la función flip. Para tablas grandes, esto es costoso, por lo que en su lugar aplicaremos la asignación generalizada a cada columna por separado, utilizando la función map (que se ve como un apóstrofo):
[aggTable;;:;]'[(idx;)each aggCols; (row[`high] | inputTable`high;row[`volume] + inputTable`volume;…)];
Volvemos a utilizar la proyección de funciones. También note que en Q crear una lista es también una función y podemos llamarla mediante la función each(map), para obtener una lista de listas.
Para que el conjunto de columnas calculadas no sea fijo, crearemos la expresión arriba de manera dinámica. Primero definiremos funciones para calcular cada columna, usando las variables row e inp para referirnos a los datos agregados y de entrada:
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");
Algunas columnas son especiales, su primer valor no debe ser calculado por la función. Podemos definir que es primero en la columna row[`numTrades] – si en ella hay 0, entonces el valor es el primero. En Q hay una función de selección — ?[Boolean list;list1;list2] – que elige un valor de la lista 1 o 2 dependiendo de la condición en el primer argumento:
// high -> ?[isFirst;inp`high;row[`high]|inp`high]
// @ - тоже обобщенное присваивание для случая когда индекс неглубокий
@[`aggExpression;specialCols;{[x;y]"?[isFirst;inp`",y,";",x,"]"};string specialCols];
Aquí llamé a la asignación generalizada con mi función (la expresión entre llaves). Se le pasa el valor actual (primer argumento) y un argumento adicional, que paso en el cuarto parámetro.
Agregaremos por separado las columnas acumulativas, ya que para ellas la función es la misma:
// volume -> row[`volume]+inp`volume
aggExpression[accumulatorCols]:{"row[`",x,"]+inp`",x } each string accumulatorCols;
Es una asignación ordinaria según los estándares de Q, solo que estoy asignando inmediatamente una lista de valores. Finalmente, crearemos la función principal:
// ":",/: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),")]}";
Con esta expresión estoy creando dinámicamente una función a partir de una cadena, que contiene la expresión que mencioné arriba. El resultado se verá así:
{[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])]}
El orden de cálculo de las columnas está invertido, ya que en Q el orden de cálculo es de derecha a izquierda.
Ahora tenemos dos funciones principales necesarias para los cálculos, solo queda agregar un poco de infraestructura y el servicio estará listo.
Pasos finales
Tenemos las funciones preprocess y updateAgg, que hacen todo el trabajo. Pero también es necesario asegurar una transición adecuada a través de los minutos y calcular los índices para la agregación. Primero definamos la función init:
init:{
tradeAgg:: 0#enlist[initWith]; // creamos una tabla tipificada vacía, enlist convierte un diccionario en una tabla, y 0# significa tomar 0 elementos de ella
currTime::00:00; // empezamos con 0, :: significa que la asignación va a una variable global
currSyms::`u#`symbol$(); // `u# convierte la lista en un árbol, para acelerar la búsqueda de elementos
offset::0; // índice en tradeAgg, donde comienza el minuto actual
rollCache:: `sym xkey update `u#sym from rollColumns#tradeAgg; // caché para los últimos valores de las columnas roll, tabla con la clave sym
}
También definiremos la función roll, que cambiará el minuto actual:
roll:{[tm]
if[currTime>tm; :init[]]; // si hemos pasado la medianoche, simplemente llamamos a init
rollCache,::offset _ rollColumns#tradeAgg; // actualizamos el caché – tomamos las columnas roll de aggTable, cortamos, insertamos en rollCache
offset::count tradeAgg;
currSyms::`u#`$();
}
Necesitaremos una función para añadir nuevos símbolos:
addSyms:{[syms]
currSyms,::syms; // añadimos a la lista de conocidos
// añadimos a la tabla sym, time y rollColumns utilizando una asignación generada.
// La función ^ inserta valores por defecto para las columnas roll, si el símbolo no está en caché. value flip table devuelve la lista de columnas en la tabla.
`tradeAgg upsert @[count[syms]#enlist initWith;`sym`time,cols rc;:;(syms;currTime), (initWith cols rc)^value flip rc:rollCache ([] sym: syms)];
}
Y, finalmente, la función upd (nombre tradicional de esta función para los servicios Q), que es llamada por el cliente para añadir datos:
upd:{[tblName;data] // tblName no lo necesitamos, pero generalmente el servicio maneja varias tablas
tm:exec distinct time from data:() xkey preprocess data; // preprocess & calc time
updMinute[data] each tm; // añadimos datos para cada minuto
};
updMinute:{[data;tm]
if[tmcurrTime; roll tm; currTime::tm]; // cambiamos el minuto, si es necesario
data:select from data where time=tm; // filtrado
if[count msyms:syms where not (syms:data`sym)in currSyms; addSyms msyms]; // nuevos símbolos
updateAgg[`tradeAgg;offset+currSyms?syms;data]; // actualizamos la tabla agregada. La función ? busca el índice de los elementos de la lista de la derecha en la lista de la izquierda.
};
Y eso es todo. Aquí está el código completo de nuestro servicio, como prometimos, solo unas pocas líneas:
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];
};
Pruebas
Comprobemos el rendimiento del servicio. Para ello, lo ejecutaremos en un proceso separado (coloque el código en el archivo service.q) y llame a la función init:
q service.q –p 5566
q)init[]
En otra consola, inicie un segundo proceso Q y conéctese al primero:
h:hopen `:host:5566
h:hopen 5566 // si ambos están en el mismo host
Primero, crearemos una lista de símbolos: 10,000 y agregaremos una función para crear una tabla aleatoria. En la segunda consola:
syms:`IBM`AAPL`GOOG,-9997?`8
rnd:{[n;t] ([] sym:n?syms; time:t+asc n#til 25; price:n?10f; size:n?10)}
Agregué al listado tres símbolos reales para que sea más fácil buscarlos en la tabla. La función rnd crea una tabla aleatoria con n filas, donde el tiempo varía de t a t+25 milisegundos.
Ahora podemos intentar enviar datos al servicio (agreguemos las primeras diez horas):
{h (`upd;`trade;rnd[10000;x])} each `time$00:00 + til 60*10
Se puede verificar en el servicio que la tabla se actualizó:
c 25 200
select from tradeAgg where sym=`AAPL
-20#select from tradeAgg where sym=`AAPL
Resultado:
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|2919Ahora realizaremos pruebas de carga para determinar cuántos datos puede procesar el servicio por minuto. Recuerde que hemos establecido un intervalo de actualización de 25 milisegundos. Por lo tanto, el servicio debe (de media) cumplir en al menos 20 milisegundos por actualización, para dar tiempo a los usuarios a solicitar los datos. Introduzca lo siguiente en el segundo proceso:
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 son dos minutos. Se puede intentar ejecutar primero para 1000 filas cada 25 milisegundos:
start 1000
En mi caso, el resultado es de alrededor de un par de milisegundos por actualización. Así que aumentaré de inmediato la cantidad de filas a 10,000:
start 10000
Resultado:
min| 00:00:00.004
avg| 9.191458
med| 9f
max| 00:00:00.030
Nada especial de nuevo, y eso que son 24 millones de filas por minuto, 400 mil por segundo. Más de 25 milisegundos de actualización solo se retrasó 5 veces, aparentemente al cambiar de minuto. Aumentemos a 100,000:
start 100000
Resultado:
min| 00:00:00.013
avg| 25.11083
med| 24f
max| 00:00:00.108
q)sum times
00:02:00.532
Como podemos ver, el servicio apenas logra mantenerse, pero aun así logra sobrevivir. Tal volumen de datos (240 millones de filas por minuto) es extremadamente grande, y en tales casos es habitual ejecutar varios clones (o incluso decenas de clones) del servicio, cada uno de los cuales procesa solo una parte de los símbolos. Sin embargo, el resultado es impresionante para un lenguaje interpretado que se centra principalmente en el almacenamiento de datos.
Puede surgir la pregunta de por qué el tiempo crece de manera no lineal con respecto al tamaño de cada actualización. La razón es que la función de compresión es efectivamente una función C que opera de manera mucho más eficiente que updateAgg. A partir de un cierto tamaño de actualización (alrededor de 10,000), updateAgg alcanza su límite y el tiempo de ejecución ya no depende del tamaño de la actualización. Gracias al paso preliminar Q, el servicio es capaz de procesar tales volúmenes de datos. Esto subraya la importancia de elegir el algoritmo correcto al trabajar con grandes volúmenes de datos. Otro aspecto importante es el almacenamiento adecuado de datos en memoria. Si los datos no se almacenaran de forma columnar o no estuvieran ordenados temporalmente, nos encontraríamos con un fenómeno conocido como TLB cache miss — la ausencia de la dirección de una página de memoria en la caché de direcciones del procesador. Buscar la dirección tarda aproximadamente 30 veces más en caso de falla, y en el caso de datos dispersos puede ralentizar el servicio varias veces.
Conclusión
En este artículo he mostrado que la base de datos KDB+ y Q son aptas no solo para el almacenamiento de grandes datos y el acceso simple a ellos a través de select, sino también para la creación de servicios de procesamiento de datos capaces de manejar cientos de millones de filas/gigabytes de datos incluso en un solo proceso Q. El lenguaje Q permite implementar algoritmos relacionados con el procesamiento de datos de manera increíblemente concisa y eficiente gracias a su naturaleza vectorial, un intérprete de dialecto SQL integrado y un conjunto de funciones de biblioteca muy efectivo.
Cabe señalar que lo expuesto anteriormente es solo una parte de las capacidades de Q, que tiene otras características únicas. Por ejemplo, un protocolo IPC extremadamente simple que borra la frontera entre procesos Q individuales y permite unir cientos de estos procesos en una única red, que puede estar distribuida en decenas de servidores en diferentes partes del mundo.
Fuente: habr.com
