Podes ler sobre que son a base de datos KDB+ e a linguaxe de programación Q, e cales son os seus puntos fortes e débiles na miña publicación anterior. e brevemente na introdución. Neste artigo, implementaremos un servizo en Q que procesará un fluxo de datos entrante e calculará varias funcións de agregación minuto a minuto en "tempo real" (é dicir, terá tempo de calcular todo antes do seguinte lote de datos). A principal característica de Q é que é unha linguaxe vectorial, o que nos permite operar non con obxectos individuais, senón con matrices deles, matrices de matrices e outros obxectos complexos. Linguaxes como Q e as súas linguaxes relacionadas, K, J e APL, son coñecidas pola súa brevidade. A miúdo, un programa que ocuparía varias pantallas de código nunha linguaxe familiar como Java pódese escribir en poucas liñas. Isto é precisamente o que quero demostrar neste artigo.
Introdución
KDB+ é unha base de datos en columnas deseñada para volumes moi grandes de datos organizados dun xeito específico (principalmente por tempo). Úsase principalmente en institucións financeiras como bancos, fondos de investimento e compañías de seguros. Q é a linguaxe interna de KDB+, que permite un traballo eficiente con estes datos. A filosofía de Q é a brevidade e a eficiencia, sacrificando a claridade. Isto débese a que unha linguaxe baseada en vectores sería difícil de entender, mentres que a brevidade e a riqueza permiten que unha parte moito maior do programa se mostre nunha única pantalla, o que en última instancia facilita a súa comprensión.
Neste artigo, implementaremos un programa completo en Q, e quizais queiras probalo. Para iso, necesitarás o propio Q. Podes descargar a versión gratuíta de 32 bits desde o sitio web de kx: Alí, se che interesa, tamén atoparás información de referencia sobre Q, un libro e diversos artigos sobre este tema.
Declaración de problemas
Hai unha fonte que envía unha táboa de datos cada 25 milisegundos. Dado que KDB+ se usa principalmente en finanzas, suporemos que é unha táboa de negociacións coas seguintes columnas: tempo (tempo en milisegundos), símbolo (símbolo da empresa na bolsa de valores – IBM, AAPL,…), prezo (o prezo ao que se compraron as accións) e tamaño (o tamaño da transacción). O intervalo de 25 milisegundos escolleuse arbitrariamente; non é nin demasiado pequeno nin demasiado grande. A súa presenza significa que os datos que chegan ao servizo xa están almacenados en búfer. Sería doado implementar o almacenamento en búfer no lado do servizo, incluíndo o almacenamento en búfer dinámico baseado na carga actual, pero para simplificar, manteremos un intervalo fixo.
O servizo debe calcular un conxunto de funcións de agregación (prezo máximo, prezo medio, tamaño da suma e outra información útil) por minuto para cada símbolo entrante da columna de símbolos. Para simplificar, asumimos que todas as funcións pódense calcular incrementalmente, o que significa que para obter un novo valor, son suficientes dous números (o valor antigo e o valor entrante). Por exemplo, as funcións máximo, medio e suma teñen esta propiedade, pero a función mediana non.
Tamén asumiremos que o fluxo de datos entrante está ordenado por tempo. Isto permitiranos traballar só co minuto máis recente. Na práctica, abonda con poder traballar cos minutos actuais e anteriores no caso de que algunha actualización sexa tardía. Para simplificar, non consideraremos este caso.
Funcións de agregación
A continuación móstranse as funcións de agregación requiridas. Incluín tantas como puiden para aumentar a carga do servizo:
- alto – prezo máximo – prezo máximo por minuto.
- baixo – prezo mínimo – prezo mínimo por minuto.
- firstPrice – first price – o primeiro prezo por minuto.
- lastPrice – último prezo – o último prezo por minuto.
- firstSize – first size – o tamaño da primeira transacción por minuto.
- lastSize – last size — o tamaño da última transacción por minuto.
- numTrades – contador i – número de operacións por minuto.
- volume – suma tamaño – suma dos tamaños das transaccións por minuto.
- pvolume – sum price – suma dos prezos por minuto, necesario para avgPrice.
- volume de negocios – suma prezo*tamaño – volume total de transaccións por minuto.
- avgPrice – pvolume%numTrades – prezo medio por minuto.
- avgSize – volume%numTrades – tamaño medio das operacións por minuto.
- vwap – facturación %volume – prezo medio por minuto ponderado polo tamaño da operación.
- cumVolume – volume sumado – volume acumulado de transaccións durante todo o período.
Vexamos de inmediato un punto non obvio: como inicializar estas columnas a primeira vez e para cada minuto posterior. Algunhas columnas, como firstPrice, deben inicializarse en nulo cada vez; o seu valor non está definido. Outras, como volume, deben establecerse sempre en 0. Tamén hai columnas que requiren unha estratexia combinada; por exemplo, cumVolume debe copiarse do minuto anterior e establecerse en 0 para o primeiro minuto. Definiremos todos estes parámetros usando o tipo de datos dicionario (semellante a un rexistro):
// 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 объяснен ниже
Engadín sym e time ao dicionario por comodidade. Agora initWith é unha fila xa feita da táboa agregada final, onde aínda hai que especificar o sym e o time correctos. Podes usalo para engadir novas filas á táboa.
Necesitaremos aggCols ao crear a función agregada. A lista debe invertirse debido á orde de avaliación da expresión en Q (de dereita a esquerda). O obxectivo é garantir que a avaliación proceda de high a cumVolume, xa que algunhas columnas dependen de columnas anteriores.
Columnas que precisan ser copiadas ao novo minuto desde o anterior, engádese a columna sym por comodidade:
rollColumns:`sym`cumVolume;
Agora imos dividir as columnas en grupos segundo como se deben actualizar. Pódense distinguir tres tipos:
- Acumuladores (volume, facturación,..): debemos sumar o valor entrante ao anterior.
- Cun punto especial (alto, baixo, ..): o primeiro valor do minuto tómase dos datos entrantes, o resto calcúlase usando a función.
- O resto calcúlase sempre coa función.
Definamos variables para estas clases:
accumulatorCols:`numTrades`volume`pvolume`turnover;
specialCols:`high`low`firstPrice`firstSize;
Orde de cálculo
Actualizaremos a táboa agregada en dous pasos. Para maior eficiencia, primeiro reduciremos a táboa de entrada para que conteña unha fila para cada símbolo e minuto. O feito de que todas as nosas funcións sexan incrementais e asociativas garante que o resultado non cambiará con este paso adicional. Poderiamos reducir a táboa usando unha sentenza select:
select high:max price, low:min price … by sym,time.minute from table
Este método ten un inconveniente: o conxunto de columnas calculadas está predefinido. Afortunadamente, Q tamén implementa select como unha función que acepta argumentos xerados dinamicamente:
?[table;whereClause;byClause;selectClause]
Non vou describir o formato do argumento en detalle; no noso caso, as únicas expresións non triviais son as expresións by e select, e deben ser dicionarios da forma columnas!expresións. Polo tanto, a función de compresión pódese definir do seguinte xeito:
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 maior claridade, empreguei a función parse, que converte unha cadea de texto que contén a expresión Q nun valor que se pode pasar á función eval e que é necesario na función select. Tamén teña en conta que o preprocesamento se define como unha proxección (é dicir, unha función con argumentos parcialmente definidos) da función select; falta un argumento (a táboa). Se aplicamos o preprocesamento a unha táboa, obtemos unha táboa comprimida.
O segundo paso é actualizar a táboa agregada. Primeiro, escribamos o 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, é común usar funcións de mapado/redución en lugar de bucles. Pero como Q é unha linguaxe vectorial e podemos aplicar con seguridade todas as operacións a todos os símbolos á vez, podemos, como primeira aproximación, eliminar o bucle por completo realizando operacións en todos os símbolos á vez:
idx:calcIdx inputTable;
row:aggTable idx;
aggTable[idx;`high]: row[`high] | inputTable`high;
aggTable[idx;`volume]: row[`volume] + inputTable`volume;
…
Pero podemos ir aínda máis lonxe. Q ten un operador único e excepcionalmente potente: o operador de asignación xeneralizada. Permite modificar un conxunto de valores nunha estrutura de datos complexa usando unha lista de índices, funcións e argumentos. No noso caso, ten este aspecto:
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, a asignación a unha táboa require unha lista de filas, non de columnas, e require a transposición da matriz (lista de columnas nunha lista de filas) usando a función flip. Para unha táboa grande, isto é caro, polo que no seu lugar, aplicamos unha asignación xeneralizada a cada columna individualmente usando a función map (que semella un apóstrofo):
.[aggTable;;:;]'[(idx;)each aggCols; (row[`high] | inputTable`high;row[`volume] + inputTable`volume;…)];
Estamos a usar de novo a proxección de funcións. Tamén tede en conta que en Q, a creación de listas tamén é unha función, e podemos chamala usando each(map) para obter unha lista de listas.
Para evitar un conxunto fixo de columnas calculadas, imos crear a expresión anterior dinamicamente. Primeiro, definamos funcións para calcular cada columna, usando as variables row e inp para facer referencia aos datos agregados e 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");
Algunhas columnas son especiais; o seu primeiro valor non debería ser calculado pola función. Podemos determinar que é o primeiro pola columna row[`numTrades]; se é 0, o valor é o primeiro. Q ten unha función de selección — ?[Boolean list;list1;list2] — que selecciona un valor da lista 1 ou 2 dependendo da condición do primeiro argumento:
// high -> ?[isFirst;inp`high;row[`high]|inp`high]
// @ - тоже обобщенное присваивание для случая когда индекс неглубокий
@[`aggExpression;specialCols;{[x;y]"?[isFirst;inp`",y,";",x,"]"};string specialCols];
Aquí chamei unha asignación xenérica coa miña función (a expresión entre chaves). Pasa o valor actual (o primeiro argumento) e un argumento adicional, que paso no cuarto parámetro.
Engadiremos os altofalantes alimentados por batería por separado, xa que cumpren a mesma función:
// volume -> row[`volume]+inp`volume
aggExpression[accumulatorCols]:{"row[`",x,"]+inp`",x } each string accumulatorCols;
Esta é unha asignación típica segundo os estándares de Q, pero estou asignando unha lista de valores á vez. Finalmente, imos crear a 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 dinamicamente unha función a partir dunha cadea de texto que contén a expresión que proporcionei anteriormente. O resultado será 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])]}
A orde de avaliación das columnas está invertida, xa que en Q a orde de avaliación é de dereita a esquerda.
Agora temos dúas funcións principais necesarias para a informática, o único que queda é engadir un pouco de infraestrutura e o servizo estará listo.
Pasos finais
Temos as funcións preprocess e updateAgg que fan todo o traballo. Pero aínda precisamos garantir que as transicións entre os minutos sexan axeitadas e calcular os índices para a agregación. Primeiro, definamos a función 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
}
Tamén definiremos unha función de rolamento que cambiará o minuto actual:
roll:{[tm]
if[currTime>tm; :init[]]; // если перевалили за полночь, то просто вызовем init
rollCache,::offset _ rollColumns#tradeAgg; // обновим кэш – взять roll колонки из aggTable, обрезать, вставить в rollCache
offset::count tradeAgg;
currSyms::`u#`$();
}
Necesitaremos unha función para engadir novos caracteres:
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)];
}
E finalmente, a función upd (o nome tradicional desta función para os servizos Q), que o cliente chama para engadir 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]; // обновим агрегированную таблицу. Функция ? ищет индекс элементов списка справа в списке слева.
};
Iso é todo. Aquí está o código completo para o noso servizo, como prometemos, só unhas poucas liñas:
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];
};
Probas
Probemos o rendemento do servizo. Para iso, iníciao nun proceso separado (coloca o código no ficheiro service.q) e chama á función init:
q service.q –p 5566
q)init[]
Noutra consola, inicia un segundo proceso Q e conéctate ao primeiro:
h:hopen `:host:5566
h:hopen 5566 // если оба на одном хосте
Primeiro, imos crear unha lista de caracteres (10000 en total) e engadir unha función para xerar unha táboa aleatoria. Na 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)}
Engadín tres símbolos reais á lista para facilitar a súa busca na táboa. A función rnd crea unha táboa aleatoria con n filas, onde os tempos van dende t ata t+25 milisegundos.
Agora podes tentar enviar datos ao servizo (engadimos as primeiras dez horas):
{h (`upd;`trade;rnd[10000;x])} each `time$00:00 + til 60*10
Podes comprobar no servizo que a táboa foi 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|2919Agora imos executar unha proba de carga para determinar cantos datos pode procesar o servizo por minuto. Como recordatorio, establecemos o intervalo de actualización en 25 milisegundos. Polo tanto, o servizo debería (de media) actualizarse en polo menos 20 milisegundos para darlles aos usuarios tempo de solicitar datos. Introduza o seguinte no 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 dous minutos. Poderías tentar executalo primeiro durante 1000 filas cada 25 milisegundos:
start 1000
No meu caso, o resultado é duns milisegundos por actualización. Polo tanto, aumentarei inmediatamente o 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
De novo, nada especial, pero iso son 24 millóns de filas por minuto, 400.000 por segundo. A actualización só se ralentizou máis de 25 milisegundos cinco veces, aparentemente debido ao cambio de minuto. Aumentemos iso 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, o servizo apenas se comporta, pero aínda así consegue manterse a flote. Este volume de datos (240 millóns de filas por minuto) é extremadamente grande; neses casos, é común lanzar varios clons (ou incluso ducias) do servizo, cada un procesando só un subconxunto dos caracteres. Non obstante, o resultado é impresionante para unha linguaxe interpretada que se centra principalmente no almacenamento de datos.
Un podería preguntarse por que o tempo medra de forma non lineal co tamaño de cada actualización. A razón é que a función de compresión é esencialmente unha función C, que é moito máis eficiente que updateAgg. A partir dun certo tamaño de actualización (arredor de 10.000), updateAgg alcanza o seu límite e, despois diso, o seu tempo de execución é independente do tamaño da actualización. É precisamente debido ao paso previo Q que o servizo é capaz de dixerir semellantes volumes de datos. Isto subliña a importancia de elixir o algoritmo correcto cando se traballa con big data. Outra consideración é o almacenamento axeitado de datos na memoria. Se os datos non se almacenasen en columnas ou ordenados no tempo, atopariamos algo chamado erro de caché TLB: un fallo ao atopar un enderezo de páxina de memoria na caché de enderezos do procesador. As buscas de enderezos tardan aproximadamente 30 veces máis se non teñen éxito e, no caso de datos dispersos, isto pode ralentizar o servizo varias veces.
Conclusión
Neste artigo, demostrei que KDB+ e Q son axeitados non só para almacenar grandes conxuntos de datos e acceder facilmente a eles mediante instrucións select, senón tamén para crear servizos de procesamento de datos capaces de procesar centos de millóns de filas/gigabytes de datos mesmo nun único proceso Q. A propia linguaxe Q permite unha implementación excepcionalmente concisa e eficiente de algoritmos de procesamento de datos debido á súa natureza vectorial, ao intérprete SQL integrado e a un conxunto moi exitoso de funcións de biblioteca.
Gustaríame sinalar que o anterior é só unha mostra das capacidades de Q; tamén ten outras características únicas. Por exemplo, un protocolo IPC extremadamente sinxelo que elimina os límites entre os procesos Q individuais e permite que centos destes procesos se conecten nunha única rede, que pode abarcar ducias de servidores en todo o mundo.
Fonte: www.habr.com
