Sobre o que é a base KDB+, a linguagem de programação Q, quais são seus pontos fortes e fracos, você pode ler no meu artigo anterior e brevemente na introdução. Neste artigo, implementaremos um serviço em Q que processará um fluxo de dados de entrada e calculará várias funções agregadas a cada minuto em tempo real (ou seja, será capaz de calcular tudo antes do próximo lote de dados). A principal característica do Q é que ele é uma linguagem vetorial, permitindo operar não apenas com objetos individuais, mas também com arrays, arrays de arrays e outros objetos complexos. Linguagens como Q e suas parentes K, J, APL são conhecidas por sua concisão. Muitas vezes, um programa que ocupa várias telas de código em uma linguagem comum como Java pode ser escrito por elas em apenas algumas linhas. É exatamente isso que quero demonstrar neste artigo.
Introdução
KDB+ é um banco de dados em coluna, projetado para manipular volumes muito grandes de dados, organizados de uma determinada maneira (principalmente por tempo). Ele é utilizado, principalmente, em instituições financeiras - bancos, fundos de investimento, companhias de seguros. A linguagem Q é a linguagem interna do KDB+, que permite trabalhar eficientemente com esses dados. A ideologia do Q é concisão e eficiência, com clareza sacrificada. Isso é justificado pelo fato de que uma linguagem vetorial será difícil de entender de qualquer forma, e a concisão e a riqueza da notação permitem ver em uma única tela uma parte muito maior do programa, o que, no final, facilita sua compreensão.
Neste artigo, implementaremos um programa completo em Q e você pode querer experimentá-lo. Para isso, você precisará do Q. Você pode baixar a versão gratuita de 32 bits no site da empresa kx - . Lá, se você estiver interessado, encontrará informações de referência sobre Q, o livro e vários artigos sobre o assunto.
Formulação do problema
Há uma fonte que envia a cada 25 milissegundos uma tabela com dados. Como o KDB+ é aplicado principalmente em finanças, vamos considerar que se trata de uma tabela de transações (trades), que possui as seguintes colunas: time (tempo em milissegundos), sym (símbolo da empresa na bolsa - IBM, AAPL,…), price (preço a que as ações foram compradas), size (tamanho do negócio). O intervalo de 25 milissegundos foi escolhido arbitrariamente, não é muito pequeno nem muito grande. Sua presença indica que os dados estão chegando ao serviço já armazenados em buffer. Seria fácil implementar a armazenagem em buffer do lado do serviço, incluindo uma versão dinâmica dependendo da carga atual, mas para simplificar, ficaremos com um intervalo fixo.
O serviço deve calcular a cada minuto para cada caractere de entrada da coluna sym um conjunto de funções agregadoras – max price, avg price, sum size e outras informações úteis. Para simplificar, assumiremos que todas as funções podem ser computadas de forma incremental, ou seja, para obter um novo valor, basta conhecer dois números – o antigo e o valor de entrada. Por exemplo, as funções max, média e soma possuem essa propriedade, enquanto a função mediana não.
Também presumimos que o fluxo de dados de entrada está ordenado por tempo. Isso nos permitirá trabalhar apenas com o último minuto. Na prática, é suficiente trabalhar com o minuto atual e o anterior, caso algumas atualizações se atrasem. Para simplificar, não consideraremos esse caso.
Funções agregadoras
Abaixo estão listadas as funções agregadoras necessárias. Eu as coletei o máximo possível para aumentar a carga sobre o serviço:
- high – max price – preço máximo por minuto.
- low – min price – preço mínimo por minuto.
- firstPrice – first price – primeiro preço por minuto.
- lastPrice – last price – último preço por minuto.
- firstSize – first size – primeiro tamanho do negócio por minuto.
- lastSize – last size – último tamanho do negócio por minuto.
- numTrades – count i – número de negócios por minuto.
- volume – sum size – soma dos tamanhos dos negócios por minuto.
- pvolume – sum price – soma dos preços por minuto, necessária para avgPrice.
- turnover – sum price*size – volume total de negócios por minuto.
- avgPrice – pvolume%numTrades – preço médio por minuto.
- avgSize – volume%numTrades – tamanho médio do negócio por minuto.
- vwap – turnover%volume – preço médio ponderado pelo tamanho do negócio por minuto.
- cumVolume – sum volume – tamanho acumulado dos negócios por todo o tempo.
Vamos discutir um ponto pouco óbvio – como inicializar estas colunas pela primeira vez e para cada minuto subsequente. Algumas colunas do tipo firstPrice precisam ser inicializadas sempre com o valor nulo, pois seu valor não está definido. Outras do tipo volume devem ser sempre definidas como 0. Existem também colunas que exigem uma abordagem combinada – por exemplo, cumVolume deve ser copiado do minuto anterior, enquanto para o primeiro, deve ser definido como 0. Vamos definir todos esses parâmetros usando o tipo de dado dicionário (análico a uma gravação):
// 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 объяснен ниже
Adicionei sym e time ao dicionário para conveniência, agora initWith é uma linha pronta da tabela agregada final, onde restam os valores corretos para sym e time. Pode-se usar isso para adicionar novas linhas à tabela.
aggCols será necessário na criação da função agregadora. A lista precisa ser invertida devido às particularidades da ordem de execução das expressões em Q (da direita para a esquerda). O objetivo é garantir o cálculo da direção do high para cumVolume, uma vez que algumas colunas dependem de valores anteriores.
Colunas que precisam ser copiadas para o novo minuto a partir do anterior, a coluna sym foi adicionada para conveniência:
rollColumns:`sym`cumVolume;
Agora vamos dividir as colunas em grupos de acordo com como devem ser atualizadas. Podemos identificar três tipos:
- Acumuladores (volume, turnover,..) – devemos adicionar o valor de entrada ao anterior.
- Com ponto específico (high, low, ..) – o primeiro valor no minuto é retirado dos dados de entrada, os demais são calculados por meio da função.
- Os demais. Sempre são calculados pela função.
Definiremos variáveis para essas classes:
accumulatorCols:`numTrades`volume`pvolume`turnover;
specialCols:`high`low`firstPrice`firstSize;
Ordem de cálculos
Atualizaremos a tabela agregada em duas etapas. Para eficiência, primeiro compactaremos a tabela de entrada para que reste uma linha para cada símbolo e minuto. O fato de que todas as nossas funções sejam incrementais e associativas garante que o resultado desse passo adicional não mudará. A tabela poderia ser compactada usando um select:
select high:max price, low:min price … by sym,time.minute from table
Esse método tem uma desvantagem – o conjunto de colunas calculadas é pré-definido. Felizmente, em Q, o select é implementado também como uma função, onde podemos passar argumentos criados dinamicamente:
?[table;whereClause;byClause;selectClause]
Não vou descrever detalhadamente o formato dos argumentos, no nosso caso apenas as expressões by e select serão não triviais e devem ser dicionários do tipo columns!expressions. Assim, a função de compressão pode ser definida assim:
selExpression:`high`low`firstPrice`lastPrice`firstSize`lastSize`numTrades`volume`pvolume`turnover!parse each ("preço máximo";"preço mínimo";"primeiro preço";"último preço";"primeiro tamanho";"último tamanho";"contagem i";"soma tamanho";"soma preço";"soma preço*tamanho"); // each é a função map em Q para uma lista única
preprocess:?[;();`sym`time!`sym`time.minute;selExpression];
Para maior clareza, utilizei a função parse, que transforma uma string com uma expressão Q em um valor que pode ser passado para a função eval e que é necessário no select funcional. Também devemos notar que preprocess é definido como uma projeção (ou seja, uma função com argumentos parcialmente definidos) da função select, onde um argumento (a tabela) está ausente. Se aplicarmos preprocess à tabela, obteremos uma tabela comprimida.
A segunda etapa é a atualização da tabela agregada. Vamos primeiro escrever o algoritmo em pseudocódigo:
for each sym in inputTable
idx: índice da linha na tabela agg para sym+currentTime;
aggTable[idx;`high]: aggTable[idx;`high] | inputTable[sym;`high];
aggTable[idx;`volume]: aggTable[idx;`volume] + inputTable[sym;`volume];
…
Em Q, é comum usar funções de map/reduce em vez de ciclos. Mas como Q é uma linguagem vetorial e todas as operações podem ser aplicadas simultaneamente a todos os símbolos, podemos, num primeiro momento, evitar completamente o ciclo, realizando operações em todos os símbolos de uma vez:
idx:calcIdx inputTable;
row:aggTable idx;
aggTable[idx;`high]: row[`high] | inputTable`high;
aggTable[idx;`volume]: row[`volume] + inputTable`volume;
…
Mas podemos ir mais longe, pois Q possui um operador único e extremamente poderoso – o operador de atribuição geral. Ele permite alterar um conjunto de valores em uma estrutura de dados complexa usando listas de índices, funções e argumentos. No nosso caso, ele se apresenta assim:
idx:calcIdx inputTable;
rows:aggTable idx;
// .[target;(idx0;idx1;..);function;argument] ~ target[idx 0;idx 1;…]: function[target[idx 0;idx 1;…];argument], no nosso caso a função é atribuição
.[aggTable;(idx;aggCols);:;flip (row[`high] | inputTable`high;row[`volume] + inputTable`volume;…)];
Infelizmente, para atribuições em uma tabela, é necessário ter uma lista de linhas, não de colunas, e é preciso transpor a matriz (converter uma lista de colunas em uma lista de linhas) usando a função flip. Para uma tabela grande, isso é oneroso, por isso aplicaremos a atribuição geral a cada coluna separadamente, usando a função map (que se apresenta como um apóstrofo):
[aggTable;;:;]'[(idx;)each aggCols; (row[`high] | inputTable`high;row[`volume] + inputTable`volume;…)];
Estamos novamente utilizando a projeção de funções. Note também que, em Q, a criação de listas também é uma função, e podemos chamá-la com a função each(map) para obter uma lista de listas.
Para que o conjunto de colunas calculadas não seja fixo, criaremos a expressão acima dinamicamente. Primeiro, definiremos funções para calcular cada coluna, utilizando as variáveis row e inp para se referir aos dados 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");
Algumas colunas são especiais, e seu primeiro valor não deve ser calculado pela função. Podemos determinar que é o primeiro pela coluna row[`numTrades] – se ela for 0, então o valor é o primeiro. Em Q, há uma função de seleção — ?[Boolean list;list1;list2] – que escolhe o valor da lista 1 ou 2 dependendo da condição no primeiro argumento:
// high -> ?[isFirst;inp`high;row[`high]|inp`high]
// @ - тоже обобщенное присваивание для случая когда индекс неглубокий
@[`aggExpression;specialCols;{[x;y]"?[isFirst;inp`",y,";",x,"]"};string specialCols];
Aqui eu chamei a atribuição genérica com minha função (a expressão entre chaves). Ela recebe o valor atual (primeiro argumento) e um argumento adicional que eu passo no quarto parâmetro.
Adicionalmente, vamos adicionar colunas acumuladoras, pois para elas a função é única:
// volume -> row[`volume]+inp`volume
aggExpression[accumulatorCols]:{"row[`",x,"]+inp`",x } each string accumulatorCols;
Esta é uma atribuição típica de Q, apenas estou atribuindo uma lista de valores de uma vez. Finalmente, criaremos a função 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),")]}";
Com esta expressão estou criando dinamicamente uma função a partir de uma string que contém a expressão que mencionei acima. O resultado ficará assim:
{[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 ordem de cálculo das colunas está invertida, pois em Q a ordem de cálculo é da direita para a esquerda.
Agora temos duas funções principais necessárias para os cálculos, resta adicionar um pouco de infraestrutura e o serviço estará pronto.
Etapas finais
Temos as funções preprocess e updateAgg, que fazem todo o trabalho. Mas ainda precisamos garantir a transição correta entre minutos e calcular os índices para agregação. Primeiramente, vamos definir a função init:
init:{
tradeAgg:: 0#enlist[initWith]; // criamos uma tabela tipificada vazia, enlist transforma o dicionário em tabela, e 0# significa pegar 0 elementos dela
currTime::00:00; // começaremos com 0, :: significa que a atribuição é em uma variável global
currSyms::`u#`symbol$(); // `u# - transforma a lista em uma árvore, para acelerar a busca de elementos
offset::0; // índice em tradeAgg, onde começa o minuto atual
rollCache:: `sym xkey update `u#sym from rollColumns#tradeAgg; // cache para os últimos valores das colunas roll, tabela com a chave sym
}
Vamos também definir a função roll, que irá mudar o minuto atual:
roll:{[tm]
if[currTime>tm; :init[]]; // se passarmos da meia-noite, apenas chamamos o init
rollCache,::offset _ rollColumns#tradeAgg; // atualizamos o cache – pegamos as colunas roll da aggTable, cortamos, inserimos no rollCache
offset::count tradeAgg;
currSyms::`u#`$();
}
Precisamos de uma função para adicionar novos símbolos:
addSyms:{[syms]
currSyms,::syms; // adicionamos à lista de conhecidos
// adicionamos à tabela sym, time e rollColumns usando atribuição genérica.
// A função ^ insere valores padrão para as colunas roll, se o símbolo não estiver no cache. value flip table retorna a lista de colunas na tabela.
`tradeAgg upsert @[count[syms]#enlist initWith;`sym`time,cols rc;:;(syms;currTime), (initWith cols rc)^value flip rc:rollCache ([] sym: syms)];
}
E, finalmente, a função upd (nome tradicional dessa função para serviços Q), que é chamada pelo cliente, para adicionar dados:
upd:{[tblName;data] // tblName não nos interessa, mas geralmente o serviço processa várias tabelas
tm:exec distinct time from data:() xkey preprocess data; // preprocess & calc time
updMinute[data] each tm; // adicionamos dados para cada minuto
};
uptMinute:{[data;tm]
if[tm<>currTime; roll tm; currTime::tm]; // mudamos o minuto, se necessário
data:select from data where time=tm; // filtragem
if[count msyms:syms where not (syms:data`sym)in currSyms; addSyms msyms]; // novos símbolos
updateAgg[`tradeAgg;offset+currSyms?syms;data]; // atualizamos a tabela agregada. A função ? encontra o índice dos elementos da lista à direita na lista à esquerda.
};
E é isso. Aqui está o código completo do nosso serviço, como prometido, apenas algumas linhas:
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];
};
Testes
Vamos verificar o desempenho do serviço. Para isso, o iniciaremos em um processo separado (coloque o código em um arquivo service.q) e chamaremos a função init:
q service.q –p 5566
q)init[]
Em outro console, inicie um segundo processo Q e conecte-se ao primeiro:
h:hopen `:host:5566
h:hopen 5566 \/\/ se ambos estiverem no mesmo host
Primeiro, criaremos uma lista de símbolos – 10000 itens e adicionaremos uma função para criar uma tabela aleatória. No segundo 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)}
Adicionei três símbolos reais à lista para facilitar a busca na tabela. A função rnd cria uma tabela aleatória com n linhas, onde o tempo varia de t a t+25 milissegundos.
Agora podemos tentar enviar dados para o serviço (adicionaremos as primeiras dez horas):
{h (`upd;`trade;rnd[10000;x])} each `time$00:00 + til 60*10
Pode-se verificar no serviço que a tabela foi atualizada:
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 vamos realizar testes de carga para descobrir quanta informação o serviço pode processar por minuto. Lembro que estabelecemos um intervalo de atualização de 25 milissegundos. Portanto, o serviço deve (em média) conseguir realizar pelo menos uma atualização em 20 milissegundos, para dar tempo aos usuários de solicitar dados. Insira o seguinte no segundo processo:
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 – são dois minutos. Podemos tentar executar primeiramente para 1000 linhas a cada 25 milissegundos:
start 1000
No meu caso, o resultado fica em torno de alguns milissegundos por atualização. Assim, vou aumentar imediatamente o número de linhas para 10.000:
start 10000
Resultado:
min| 00:00:00.004
avg| 9.191458
med| 9f
max| 00:00:00.030
Nada impressionante novamente, considerando que isso equivale a 24 milhões de linhas por minuto, 400 mil por segundo. A atualização demorou mais de 25 milissegundos apenas 5 vezes, aparentemente durante a troca de minutos. Vamos aumentar para 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 serviço mal consegue se manter, mas, ainda assim, consegue se manter à tona. Esse volume de dados (240 milhões de linhas por minuto) é extremamente alto; nesses casos, geralmente se inicia várias instâncias (ou até dezenas de instâncias) do serviço, cada uma processando apenas uma parte dos símbolos. No entanto, o resultado é impressionante para uma linguagem interpretada, que se concentra principalmente no armazenamento de dados.
Pode surgir a pergunta sobre por que o tempo aumenta de forma não linear com o tamanho de cada atualização. A razão é que a função de compressão é, na verdade, uma função C, que opera de maneira muito mais eficiente que o updateAgg. A partir de um certo tamanho de atualização (cerca de 10.000), o updateAgg atinge seu limite, e seu tempo de execução não depende mais do tamanho da atualização. É justamente por meio do passo preliminar Q que o serviço é capaz de processar tais volumes de dados. Isso ressalta a importância de escolher o algoritmo correto ao trabalhar com big data. Outro ponto é a correta armazenagem de dados na memória. Se os dados não fossem armazenados de forma colunar ou não estivessem ordenados temporalmente, nos depararíamos com um fenômeno conhecido como TLB cache miss – a falta de um endereço de página de memória no cache de endereços do processador. A busca pelo endereço leva cerca de 30 vezes mais tempo em caso de falha e, em situações de dados dispersos, pode retardar o serviço várias vezes.
Conclusão
Neste artigo, demonstrei que a base KDB+ e Q são não apenas adequadas para armazenar grandes volumes de dados e acessá-los facilmente por meio de selects, mas também para criar serviços de processamento de dados que conseguem gerenciar centenas de milhões de linhas/ gigabytes de dados, mesmo em um único processo Q. A própria linguagem Q permite implementar algoritmos relacionados ao processamento de dados de forma excepcionalmente concisa e eficaz, graças à sua natureza vetorial, ao interpretador embutido de SQL e a um conjunto muito bem elaborado de funções de biblioteca.
Vale ressaltar que o acima exposto é apenas uma parte das capacidades do Q; ele possui outras características únicas. Por exemplo, um protocolo IPC extremamente simples que elimina a fronteira entre processos Q individuais e permite a união de centenas desses processos em uma única rede que pode estar distribuída em dezenas de servidores em diferentes partes do mundo.
Fonte: habr.com
