Qu'est-ce que KDB +, le langage de programmation Q, quelles sont leurs forces et leurs faiblesses, peut être trouvé dans mon
article précédent et brièvement dans l'introduction. Dans l'article, nous implémentons un service sur Q qui traitera le flux de données entrant et calculera par minute diverses fonctions d'agrégation en mode «temps réel» (c'est-à-dire qu'il parviendra à tout compter jusqu'à la prochaine donnée). La principale caractéristique de Q est qu'il s'agit d'un langage vectoriel qui vous permet de fonctionner non pas avec des objets uniques, mais avec leurs tableaux, tableaux de tableaux et autres objets complexes. Des langages tels que Q et ses K, J, APL sont réputés pour leur brièveté. Souvent, un programme qui s'étend sur plusieurs écrans de code dans un langage familier tel que Java peut y être écrit sur plusieurs lignes. C'est exactement ce que je veux démontrer dans cet article.
Présentation
KDB + est une base de données de colonnes centrée sur de très gros volumes de données, triées d'une certaine manière (principalement par le temps). Il est utilisé, tout d'abord, dans les organisations financières - banques, fonds d'investissement, compagnies d'assurance. Le langage Q est un langage interne de KDB + qui vous permet de travailler efficacement avec ces données. L'idéologie de Q est la brièveté et l'efficacité, tandis que la clarté est sacrifiée. Cela se justifie par le fait que dans tous les cas, le langage vectoriel sera difficile à percevoir, et la brièveté et la richesse de l'enregistrement vous permettent de voir une partie beaucoup plus grande du programme sur un seul écran, ce qui facilite finalement sa compréhension.
Dans cet article, nous mettons en œuvre un programme Q à part entière et vous voudrez peut-être l'essayer. Pour ce faire, vous aurez besoin du véritable Q. Vous pouvez télécharger la version gratuite 32 bits sur le site Web de la société kx -
www.kx.com . Au même endroit, si vous êtes intéressé, vous trouverez des informations de référence sur Q, le livre
Q For Mortals et divers articles sur ce sujet.
Énoncé du problème
Il existe une source qui envoie une table de données toutes les 25 millisecondes. Étant donné que KDB + est principalement utilisé dans la finance, nous supposons qu'il s'agit d'une table de transactions dans laquelle se trouvent les colonnes suivantes: heure (heure en millisecondes), sym (nom de la société en bourse -
IBM ,
AAPL , ...), prix (prix par lequel les actions ont été achetées), taille (taille de la transaction). Un intervalle de 25 millisecondes est choisi arbitrairement, il n'est ni trop petit ni trop grand. Sa présence signifie que les données arrivant au service sont déjà tamponnées. Il serait facile d'implémenter la mise en mémoire tampon côté service, y compris la mise en mémoire tampon dynamique, en fonction de la charge actuelle, mais pour plus de simplicité, nous nous attardons sur un intervalle fixe.
Le service doit considérer par minute pour chaque caractère entrant de la colonne sym un ensemble de fonctions d'agrégation - prix max, prix moyen, taille de la somme, etc. informations utiles. Pour simplifier, nous supposons que toutes les fonctions peuvent être calculées de manière incrémentielle, c'est-à-dire pour obtenir une nouvelle valeur, il suffit de connaître deux nombres - l'ancien et la valeur entrante. Par exemple, les fonctions max, average, sum ont cette propriété, mais pas la fonction médiane.
Nous supposons également que le flux de données entrant est ordonné par le temps. Cela nous donnera l'occasion de travailler uniquement avec la dernière minute. En pratique, il suffit de pouvoir travailler avec les minutes actuelles et précédentes au cas où des mises à jour seraient en retard. Par souci de simplicité, nous ne considérerons pas ce cas.
Fonctions d'agrégation
Voici les fonctions d'agrégation requises. Je les ai pris le plus possible pour augmenter la charge sur le service:
- haut - prix maximum - prix maximum par minute.
- low - min price - le prix minimum par minute.
- firstPrice - premier prix - le premier prix par minute.
- lastPrice - dernier prix - le dernier prix par minute.
- firstSize - first size - la taille de la première offre en une minute.
- lastSize - last size - la dernière taille de transaction en une minute.
- numTrades - count i - le nombre de transactions par minute.
- volume - somme taille - somme des tailles de transaction par minute.
- pvolume - somme prix - la somme des prix par minute, nécessaire pour le prix moyen.
- chiffre d'affaires - prix total * taille - volume total de transactions par minute.
- avgPrice - pvolume% numTrades - prix moyen par minute.
- avgSize - volume% numTrades - taille moyenne des transactions par minute.
- vwap - chiffre d'affaires% volume - le prix moyen par minute pondéré par la taille de la transaction.
- cumVolume - somme du volume - taille cumulée des transactions pendant tout le temps.
Discutez immédiatement d'un point non évident - comment initialiser ces colonnes pour la première fois et pour chaque minute suivante. Certaines colonnes du type firstPrice doivent être initialisées avec null à chaque fois, leur valeur n'est pas définie. Les autres types de volume doivent toujours être définis sur 0. Il y a encore des colonnes qui nécessitent une approche combinée - par exemple, cumVolume doit être copié à partir de la minute précédente, et pour le premier ensemble sur 0. Nous définirons tous ces paramètres à l'aide du dictionnaire de types de données (analogue de l'enregistrement):
J'ai ajouté sym et time au dictionnaire pour plus de commodité, maintenant initWith est une ligne finie de la table agrégée finale, où il reste à définir les sym et time corrects. Vous pouvez l'utiliser pour ajouter de nouvelles lignes à la table.
aggCols dont nous avons besoin lors de la création d'une fonction d'agrégation. La liste doit être inversée en raison des particularités de l'ordre dans lequel les expressions sont calculées en Q (de droite à gauche). Le but est de fournir un calcul dans la direction de high à cumVolume, car certaines colonnes dépendent des précédentes.
Colonnes à copier à une nouvelle minute de la précédente, colonne sym ajoutée pour plus de commodité:
rollColumns:`sym`cumVolume;
Maintenant, nous divisons les colonnes en groupes selon la façon dont elles doivent être mises à jour. On peut distinguer trois types:
- Batteries (volume, chiffre d'affaires, ..) - nous devons ajouter la valeur d'entrée à la précédente.
- Avec un point spécial (haut, bas, ..) - la première valeur en une minute est prise à partir des données d'entrée, les autres sont comptées à l'aide de la fonction.
- Le reste. Toujours compté à l'aide d'une fonction.
Définissez des variables pour ces classes:
accumulatorCols:`numTrades`volume`pvolume`turnover; specialCols:`high`low`firstPrice`firstSize;
Ordre de calcul
Nous mettrons à jour le tableau agrégé en deux étapes. Pour plus d'efficacité, nous allons d'abord réduire la table entrante afin qu'il ne reste qu'une ligne pour chaque caractère et chaque minute. Le fait que toutes nos fonctions soient incrémentales et associatives nous garantit que le résultat de cette étape supplémentaire ne changera pas. Vous pouvez serrer la table en utilisant la sélection:
select high:max price, low:min price … by sym,time.minute from table
Cette méthode a un inconvénient - l'ensemble des colonnes calculées est prédéfini. Heureusement, dans Q, la sélection est également implémentée comme une fonction où vous pouvez remplacer des arguments créés dynamiquement:
?[table;whereClause;byClause;selectClause]
Je ne décrirai pas en détail le format des arguments, dans notre cas, seules les expressions by et select sont non triviales et doivent être des dictionnaires des colonnes du formulaire! Expressions. Ainsi, la fonction de restriction peut être définie comme suit:
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");
Pour plus de clarté, j'ai utilisé la fonction d'analyse, qui transforme une chaîne avec une expression Q en une valeur qui peut être transmise à la fonction eval et qui est requise dans la fonction select. Notez également que le prétraitement est défini comme une projection (c'est-à-dire une fonction avec des arguments partiellement définis) de la fonction de sélection, un argument (table) est manquant. Si nous appliquons un prétraitement à une table, nous obtenons une table réduite.
La deuxième étape consiste à mettre à jour la table agrégée. Tout d'abord, nous écrivons l'algorithme en pseudocode:
for each sym in inputTable idx: row index in agg table for sym+currentTime; aggTable[idx;`high]: aggTable[idx;`high] | inputTable[sym;`high]; aggTable[idx;`volume]: aggTable[idx;`volume] + inputTable[sym;`volume]; …
Dans Q, au lieu de boucles, il est habituel d'utiliser des fonctions de cartographie / réduction. Mais comme Q est un langage vectoriel et que toutes les opérations peuvent être appliquées en toute sécurité à tous les symboles à la fois, alors en première approximation, nous pouvons nous passer d'un cycle, en effectuant des opérations avec tous les symboles à la fois:
idx:calcIdx inputTable; row:aggTable idx; aggTable[idx;`high]: row[`high] | inputTable`high; aggTable[idx;`volume]: row[`volume] + inputTable`volume; …
Mais nous pouvons aller plus loin, dans Q il y a un opérateur unique et extrêmement puissant - l'opérateur d'affectation généralisé. Il vous permet de modifier l'ensemble de valeurs dans une structure de données complexe à l'aide d'une liste d'index, de fonctions et d'arguments. Dans notre cas, cela ressemble à ceci:
idx:calcIdx inputTable; rows:aggTable idx;
Malheureusement, pour affecter à une table, vous avez besoin d'une liste de lignes, pas de colonnes, et vous devez transposer la matrice (liste de colonnes en une liste de lignes) à l'aide de la fonction flip. Pour une grande table, cela n'est pas nécessaire, donc nous appliquons à la place l'affectation généralisée à chaque colonne séparément, en utilisant la fonction de carte (qui ressemble à une apostrophe):
.[aggTable;;:;]'[(idx;)each aggCols; (row[`high] | inputTable`high;row[`volume] + inputTable`volume;…)];
Nous utilisons à nouveau la fonction de projection. Notez également que dans Q, la création d'une liste est également une fonction et nous pouvons l'appeler en utilisant la fonction each (map) pour obtenir une liste de listes.
Pour que l'ensemble des colonnes calculées ne soit pas fixe, créez dynamiquement l'expression ci-dessus. Tout d'abord, nous définissons les fonctions de calcul de chaque colonne, en utilisant les variables ligne et inp pour référencer les données agrégées et d'entrée:
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");
Certaines colonnes sont spéciales, leur première valeur ne doit pas être calculée par une fonction. Nous pouvons déterminer qu'il est le premier de la ligne de colonne [`numTrades] - s'il a 0, alors la valeur est la première. Q a une fonction de sélection -? [Liste booléenne; liste1; liste2] - qui sélectionne une valeur dans la liste 1 ou 2 selon la condition du premier argument:
Ici, j'ai appelé une affectation générique avec ma fonction (expression entre accolades). La valeur actuelle (le premier argument) et un argument supplémentaire, que je passe dans le 4ème paramètre, lui sont passés.
Séparément, nous ajoutons des haut-parleurs à batterie, car pour eux la fonction est la même:
Il s'agit d'une affectation habituelle selon les normes de Q, je n'attribue qu'une liste de valeurs à la fois. Enfin, créez la fonction principale:
Avec cette expression, je crée dynamiquement une fonction à partir d'une chaîne qui contient l'expression que j'ai citée ci-dessus. Le résultat ressemblera à ceci:
{[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])]}
L'ordre de calcul des colonnes est inversé, car dans Q l'ordre de calcul est de droite à gauche.
Maintenant, nous avons deux fonctions principales nécessaires pour les calculs, il reste à ajouter un peu d'infrastructure et le service est prêt.
Étapes finales
Nous avons des fonctions de prétraitement et de mise à jour qui font tout le travail. Mais il reste nécessaire d'assurer la transition correcte en quelques minutes et de calculer les indices d'agrégation. Nous définissons d'abord la fonction init:
init:{ tradeAgg:: 0#enlist[initWith];
Nous définissons également la fonction roll, qui changera la minute courante:
roll:{[tm] if[currTime>tm; :init[]];
Nous avons besoin d'une fonction pour ajouter de nouveaux personnages:
addSyms:{[syms] currSyms,::syms;
Et enfin, la fonction upd (le nom traditionnel de cette fonction pour les services Q), qui est appelée par le client, pour ajouter des données:
upd:{[tblName;data]
C’est tout. Voici le code complet de notre service, comme promis, en quelques lignes:
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]; };
Test
Vérifiez les performances du service. Pour ce faire, exécutez-le dans un processus distinct (placez le code dans le fichier service.q) et appelez la fonction init:
q service.q –p 5566 q)init[]
Dans une autre console, démarrez le deuxième processus Q et connectez-vous au premier:
h:hopen `:host:5566 h:hopen 5566
Tout d'abord, créez une liste de caractères - 10 000 pièces et ajoutez une fonction pour créer une table aléatoire. Dans la deuxième 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)}
J'ai ajouté trois vrais personnages à la liste des personnages pour le rendre plus pratique à rechercher dans le tableau. La fonction rnd crée une table aléatoire avec n lignes, où le temps varie de t à t + 25 millisecondes.
Vous pouvez maintenant essayer d'envoyer des données au service (ajoutez les dix premières heures):
{h (`upd;`trade;rnd[10000;x])} each `time$00:00 + til 60*10
Vous pouvez vérifier dans le service que la table a été mise à jour:
\c 25 200 select from tradeAgg where sym=`AAPL -20#select from tradeAgg where sym=`AAPL
Résultat:
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|2919
Nous allons maintenant effectuer des tests de charge pour déterminer la quantité de données que le service peut traiter par minute. Permettez-moi de vous rappeler que nous avons défini l'intervalle de mise à jour à 25 millisecondes. Par conséquent, un service devrait (en moyenne) tenir dans au moins 20 millisecondes par mise à jour pour donner aux utilisateurs le temps de demander des données. Saisissez les informations suivantes dans le deuxième processus:
tm:10:00:00.000 stressTest:{[n] 1 string[tm]," "; times,::h ({st:.zT; upd[`trade;x]; .zT-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 est deux minutes. Vous pouvez essayer de commencer en premier pour 1000 lignes toutes les 25 millisecondes:
start 1000
Dans mon cas, le résultat est d'environ deux millisecondes par mise à jour. Je vais donc immédiatement augmenter le nombre de lignes à 10 000:
start 10000
Résultat:
min| 00:00:00.004 avg| 9.191458 med| 9f max| 00:00:00.030
Encore une fois, rien de spécial, mais c'est 24 millions de lignes par minute, 400 000 par seconde. Pendant plus de 25 millisecondes, la mise à jour n'a ralenti que 5 fois, apparemment lors du changement de minute. Augmentation à 100 000:
start 100000
Résultat:
min| 00:00:00.013 avg| 25.11083 med| 24f max| 00:00:00.108 q)sum times 00:02:00.532
Comme vous pouvez le voir, le service fait à peine face, mais il parvient néanmoins à rester à flot. Cette quantité de données (240 millions de lignes par minute) est extrêmement importante, dans de tels cas, il est habituel d'exécuter plusieurs clones (voire des dizaines de clones) du service, chacun ne traitant qu'une partie des caractères. Néanmoins, le résultat est impressionnant pour le langage interprété, qui se concentre principalement sur le stockage de données.
La question peut se poser, pourquoi le temps augmente de façon non linéaire avec la taille de chaque mise à jour. La raison en est que la fonction de compression est en fait une fonction C qui fonctionne beaucoup plus efficacement que updateAgg. À partir d'une certaine taille de mise à jour (environ 10 000), updateAgg atteint son plafond, puis son temps d'exécution ne dépend pas de la taille de la mise à jour. C'est grâce à l'étape préliminaire Q que le service est capable de digérer de tels volumes de données. Cela souligne combien il est important lorsque vous travaillez avec des mégadonnées de choisir le bon algorithme. Un autre point est le stockage correct des données en mémoire. Si les données n'étaient pas stockées dans des colonnes ou n'étaient pas triées dans le temps, nous nous familiariserions avec une chose telle que le cache TLB manqué - l'absence d'une adresse de page de mémoire dans le cache d'adresse du processeur. La recherche de l'adresse prend environ 30 fois plus de temps en cas d'échec et dans le cas de données dispersées peut ralentir le service plusieurs fois.
Conclusion
Dans cet article, j'ai montré que les bases de données KDB + et Q conviennent non seulement pour stocker des données volumineuses et un accès facile à celles-ci via select, mais également pour créer des services de traitement des données qui peuvent digérer des centaines de millions de lignes / gigaoctets de données même en un seul processus Q . Le langage Q lui-même permet de mettre en œuvre extrêmement brièvement et efficacement des algorithmes liés au traitement des données en raison de sa nature vectorielle, de l'interpréteur intégré du dialecte SQL et d'un ensemble très réussi de fonctions de bibliothèque.
Je remarquerai que ce qui précède n'est qu'une partie des capacités de Q, il a d'autres caractéristiques uniques. Par exemple, un protocole IPC extrêmement simple qui efface la frontière entre les processus Q individuels et vous permet de combiner des centaines de ces processus en un seul réseau, qui peut être situé sur des dizaines de serveurs dans différentes parties du monde.