Puedes leer sobre qué es la base KDB+, el lenguaje de programación Q, cuáles son sus fortalezas y debilidades en mi artículo anterior. y brevemente en la introducción. En el artículo, implementaremos un servicio en Q que procesará el flujo de datos entrante y calculará varias funciones de agregación cada minuto en modo "tiempo real" (es decir, tendrá tiempo de calcular todo antes de la siguiente porción de datos). La característica principal de Q es que es un lenguaje vectorial que le permite operar no con objetos individuales, sino con sus matrices, matrices de matrices y otros objetos complejos. Idiomas como Q y sus parientes K, J, APL son famosos por su brevedad. A menudo, un programa que ocupa varias pantallas de código en un lenguaje familiar como Java se puede escribir en ellas en unas pocas líneas. Esto es lo que quiero demostrar en este artículo.
introducción
KDB+ es una base de datos en columnas centrada en cantidades muy grandes de datos, ordenados de una manera específica (principalmente por tiempo). Se utiliza principalmente en instituciones financieras: bancos, fondos de inversión, compañías de seguros. El lenguaje Q es el lenguaje interno de KDB+ que le permite trabajar eficazmente con estos datos. La ideología Q es brevedad y eficiencia, mientras se sacrifica la claridad. Esto se justifica por el hecho de que el lenguaje vectorial será difícil de entender en cualquier caso, y la brevedad y riqueza de la grabación permite ver una parte mucho mayor del programa en una pantalla, lo que finalmente hace que sea más fácil de entender.
En este artículo implementamos un programa completo en Q y quizás quieras probarlo. Para hacer esto, necesitará la Q real. Puede descargar la versión gratuita de 32 bits en el sitio web de la empresa kx: . Allí, si estás interesado, encontrarás información de referencia sobre Q, el libro y varios artículos sobre este tema.
Formulación del problema
Hay una fuente que envía una tabla con datos cada 25 milisegundos. Dado que KDB+ se utiliza principalmente en finanzas, asumiremos que se trata de una tabla de transacciones (trades), que tiene las siguientes columnas: time (tiempo en milisegundos), sym (designación de la empresa en la bolsa de valores). IBM, AAPL,…), precio (el precio al que se compraron las acciones), tamaño (tamaño de la transacción). El intervalo de 25 milisegundos es arbitrario, ni demasiado pequeño ni demasiado largo. Su presencia significa que los datos llegan al servicio ya almacenados en el buffer. Sería fácil implementar el almacenamiento en búfer en el lado del servicio, incluido el almacenamiento en búfer dinámico dependiendo de la carga actual, pero para simplificar, nos centraremos en un intervalo fijo.
El servicio debe contar cada minuto para cada símbolo entrante de la columna sym con un conjunto de funciones de agregación: precio máximo, precio promedio, tamaño de la suma, etc. información útil. Para simplificar, asumiremos que todas las funciones se pueden calcular de forma incremental, es decir, para obtener un nuevo valor, basta con conocer dos números: el valor antiguo y el entrante. Por ejemplo, las funciones máx, promedio y suma tienen esta propiedad, pero la función mediana no.
También asumiremos que el flujo de datos entrante está ordenado en el tiempo. Esto nos dará la oportunidad de trabajar sólo con el último minuto. En la práctica, basta con poder trabajar con los minutos actuales y anteriores en caso de que algunas actualizaciones lleguen tarde. Por simplicidad, no consideraremos este caso.
Funciones de agregación
Las funciones de agregación requeridas se enumeran a continuación. Tomé tantos como sea posible para aumentar la carga del servicio:
- alto – precio máximo – precio máximo por minuto.
- bajo – precio mínimo – precio mínimo por minuto.
- firstPrice – primer precio – primer precio por minuto.
- lastPrice – último precio – último precio por minuto.
- firstSize – primer tamaño – primer tamaño comercial por minuto.
- lastSize – último tamaño – último tamaño comercial en un minuto.
- numTrades – cuenta i – número de operaciones por minuto.
- volumen – tamaño de la suma – suma de los tamaños de las operaciones por minuto.
- pvolume – precio suma – suma de precios por minuto, requerida para avgPrice.
- – suma del precio de facturación*tamaño – volumen total de transacciones por minuto.
- avgPrice – pvolume%numTrades – precio promedio por minuto.
- avgSize – volumen%numTrades – tamaño promedio de operación por minuto.
- vwap – %volumen de facturación – precio medio por minuto ponderado por el tamaño de la transacción.
- cumVolume – volumen total – tamaño acumulado de transacciones durante todo el tiempo.
Analicemos de inmediato un punto no obvio: cómo inicializar estas columnas por primera vez y para cada minuto posterior. Algunas columnas del tipo firstPrice deben inicializarse a nulo cada vez; su valor no está definido. Otros tipos de volúmenes siempre deben establecerse en 0. También hay columnas que requieren un enfoque combinado; por ejemplo, cumVolume debe copiarse del minuto anterior y, para el primero, establecerse en 0. Configuremos todos estos parámetros usando los datos del diccionario. tipo (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 объяснен ниже
Agregué sym y time al diccionario por conveniencia, ahora initWith es una línea lista para usar de la tabla agregada final, donde queda por configurar el sym y el tiempo correctos. Puede usarlo para agregar nuevas filas a una tabla.
Necesitaremos aggCols al crear una función de agregación. La lista debe invertirse debido al orden en que se evalúan las expresiones en Q (de derecha a izquierda). El objetivo es asegurar que el cálculo vaya de alto a cumVolume, ya que algunas columnas dependen de las anteriores.
Las columnas que deben copiarse en un nuevo minuto del anterior, se agrega la columna sym por conveniencia:
rollColumns:`sym`cumVolume;
Ahora dividamos las columnas en grupos según cómo deben actualizarse. Se pueden distinguir tres tipos:
- Acumuladores (volumen, facturación,..) – debemos sumar el valor entrante al anterior.
- Con un punto especial (alto, bajo, ...): el primer valor del minuto se toma de los datos entrantes, el resto se calcula utilizando la función.
- Descansar. Siempre calculado mediante una función.
Definamos variables para estas clases:
accumulatorCols:`numTrades`volume`pvolume`turnover;
specialCols:`high`low`firstPrice`firstSize;
Orden de cálculo
Actualizaremos la tabla agregada en dos etapas. Para mayor eficiencia, primero reducimos la tabla entrante para que solo haya una fila para cada carácter y minuto. El hecho de que todas nuestras funciones sean incrementales y asociativas garantiza que el resultado de este paso adicional no cambiará. Podrías reducir la tabla usando select:
select high:max price, low:min price … by sym,time.minute from table
Este método tiene una desventaja: el conjunto de columnas calculadas está predefinido. Afortunadamente, en Q, select también se implementa como una función donde puedes sustituir argumentos creados dinámicamente:
?[table;whereClause;byClause;selectClause]
No describiré en detalle el formato de los argumentos; en nuestro caso, solo las expresiones by y select no serán triviales y deben ser diccionarios en forma de columnas!expresiones. Por tanto, la función de contracción se puede definir 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 это функция map в Q для одного списка
preprocess:?[;();`sym`time!`sym`time.minute;selExpression];
Para mayor claridad, utilicé la función de análisis, que convierte una cadena con una expresión Q en un valor que se puede pasar a la función de evaluación y que es necesario en la función de selección. También tenga en cuenta que el preproceso se define como una proyección (es decir, una función con argumentos parcialmente definidos) de la función de selección; falta un argumento (la tabla). Si aplicamos el preproceso a una tabla, obtendremos una tabla comprimida.
La segunda etapa es actualizar la tabla agregada. Primero escribamos 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 usar funciones de mapa/reducción en lugar de bucles. Pero dado que Q es un lenguaje vectorial y podemos aplicar fácilmente todas las operaciones a todos los símbolos a la vez, entonces, en una primera aproximación, podemos prescindir de un bucle, realizando operaciones en todos los símbolos a la vez:
idx:calcIdx inputTable;
row:aggTable idx;
aggTable[idx;`high]: row[`high] | inputTable`high;
aggTable[idx;`volume]: row[`volume] + inputTable`volume;
…
Pero podemos ir más allá: Q tiene un operador único y extremadamente poderoso: el operador de asignación generalizada. Le permite cambiar 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], в нашем случае функция – это присваивание
.[aggTable;(idx;aggCols);:;flip (row[`high] | inputTable`high;row[`volume] + inputTable`volume;…)];
Desafortunadamente, para asignar a una tabla necesita una lista de filas, no de columnas, y debe transponer la matriz (lista de columnas a lista de filas) usando la función voltear. Esto es costoso para una tabla grande, por lo que aplicamos una asignación generalizada a cada columna por separado, usando la función de mapa (que parece un apóstrofe):
.[aggTable;;:;]'[(idx;)each aggCols; (row[`high] | inputTable`high;row[`volume] + inputTable`volume;…)];
Nuevamente utilizamos la proyección de funciones. También tenga en cuenta que en Q, crear una lista también es una función y podemos llamarla usando la función each(map) para obtener una lista de listas.
Para garantizar que el conjunto de columnas calculadas no sea fijo, crearemos la expresión anterior de forma dinámica. Primero definamos funciones para calcular cada columna, usando las variables fila e inp para hacer referencia 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 determinar que es el primero mediante la columna fila[`numTrades]; si contiene 0, entonces el valor es el primero. Q tiene una función de selección - ?[Boolean list;list1;list2] - que selecciona 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 una tarea generalizada con mi función (una expresión entre llaves). Recibe el valor actual (el primer argumento) y un argumento adicional, que paso en el cuarto parámetro.
Agreguemos parlantes con batería por separado, ya que la función es la misma para ellos:
// volume -> row[`volume]+inp`volume
aggExpression[accumulatorCols]:{"row[`",x,"]+inp`",x } each string accumulatorCols;
Esta es una asignación normal según los estándares Q, pero asigno una lista de valores a la vez. Finalmente, creemos 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, creo dinámicamente una función a partir de una cadena que contiene la expresión que proporcioné anteriormente. 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 evaluación de las columnas se invierte porque en Q el orden de evaluación es de derecha a izquierda.
Ahora tenemos dos funciones principales necesarias para los cálculos, solo necesitamos agregar un poco de infraestructura y el servicio estará listo.
Pasos finales
Tenemos funciones de preprocesamiento y actualizaciónAgg que hacen todo el trabajo. Pero aún es necesario garantizar la correcta transición entre las actas y calcular los índices para la agregación. Primero que nada, definamos la función de inicio:
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
}
También definiremos la función roll, que cambiará el minuto actual:
roll:{[tm]
if[currTime>tm; :init[]]; // если перевалили за полночь, то просто вызовем init
rollCache,::offset _ rollColumns#tradeAgg; // обновим кэш – взять roll колонки из aggTable, обрезать, вставить в rollCache
offset::count tradeAgg;
currSyms::`u#`$();
}
Necesitaremos una función para agregar nuevos personajes:
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)];
}
Y finalmente, la función upd (el nombre tradicional de esta función para servicios Q), que es llamada por el cliente para agregar datos:
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]; // обновим агрегированную таблицу. Функция ? ищет индекс элементов списка справа в списке слева.
};
Eso es todo. Aquí está el código completo de nuestro servicio, como prometimos, sólo 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 hacer esto, ejecútelo en un proceso separado (ponga 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, inicia el segundo proceso Q y conéctate al primero:
h:hopen `:host:5566
h:hopen 5566 // если оба на одном хосте
Primero, creemos una lista de símbolos: 10000 piezas y agreguemos 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é tres símbolos reales a la lista 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 puedes intentar enviar datos al servicio (suma las primeras diez horas):
{h (`upd;`trade;rnd[10000;x])} each `time$00:00 + til 60*10
Puedes comprobar en el servicio que la tabla ha sido actualizada:
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 realicemos pruebas de carga para averiguar cuántos datos puede procesar el servicio por minuto. Permítanme recordarles que configuramos el intervalo de actualización en 25 milisegundos. En consecuencia, el servicio debe (en promedio) adaptarse a al menos 20 milisegundos por actualización para dar tiempo a los usuarios a solicitar datos. Ingrese 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. Puedes intentar ejecutar primero 1000 filas cada 25 milisegundos:
start 1000
En mi caso, el resultado ronda un par de milisegundos por actualización. Así que inmediatamente aumentaré el número de filas a 10.000:
start 10000
Resultado:
min| 00:00:00.004
avg| 9.191458
med| 9f
max| 00:00:00.030
Una vez más, nada especial, pero son 24 millones de líneas por minuto, 400 mil por segundo. Durante más de 25 milisegundos, la actualización se ralentizó sólo 5 veces, aparentemente cuando cambió el 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 puede ver, el servicio apenas da abasto, pero aun así logra mantenerse a flote. Semejante volumen de datos (240 millones de líneas por minuto) es extremadamente grande; en tales casos, es habitual lanzar varios clones (o incluso decenas de clones) del servicio, cada uno de los cuales procesa sólo una parte de los caracteres. Aun así, 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 forma no lineal con el tamaño de cada actualización. La razón es que la función de reducción es en realidad una función de C, que es 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 luego su tiempo de ejecución no depende del tamaño de la actualización. Gracias al paso preliminar Q, el servicio puede digerir tales volúmenes de datos. Esto resalta lo importante que es elegir el algoritmo correcto cuando se trabaja con big data. Otro punto es el correcto almacenamiento de datos en la memoria. Si los datos no se almacenaran en columnas o no estuvieran ordenados por tiempo, entonces nos familiarizaríamos con un error de caché TLB: la ausencia de una dirección de página de memoria en el caché de direcciones del procesador. La búsqueda de una dirección tarda unas 30 veces más si no tiene éxito, y si los datos están dispersos, el servicio puede ralentizarse varias veces.
Conclusión
En este artículo, mostré que las bases de datos KDB+ y Q son adecuadas no sólo para almacenar grandes datos y acceder fácilmente a ellos a través de selección, sino también para crear servicios de procesamiento de datos que son capaces de digerir cientos de millones de filas/gigabytes de datos incluso en un solo proceso Q. El lenguaje Q en sí permite una implementación extremadamente concisa y eficiente de algoritmos relacionados con el procesamiento de datos debido a su naturaleza vectorial, su intérprete de dialecto SQL incorporado y un conjunto muy exitoso de funciones de biblioteca.
Señalaré que lo anterior es solo una parte de lo que Q puede hacer; también tiene otras características únicas. Por ejemplo, un protocolo IPC extremadamente simple que borra la frontera entre procesos Q individuales y le permite combinar cientos de estos procesos en una sola red, que puede ubicarse en docenas de servidores en diferentes partes del mundo.
Fuente: habr.com