Über die KDB+-Datenbank, die Programmiersprache Q und ihre Stärken und Schwächen können Sie in meinem vorherigen Artikel nachlesen und kurz in der Einleitung. In diesem Artikel werden wir einen Q-Service implementieren, der einen eingehenden Datenstrom verarbeitet und verschiedene aggregierte Funktionen in „Echtzeit“ berechnet (das heißt, er wird alle Berechnungen vor der nächsten Datenportion abschließen). Das Hauptmerkmal von Q ist, dass es sich um eine vektorbasierte Sprache handelt, die es ermöglicht, nicht nur mit einzelnen Objekten, sondern auch mit Arrays, Arrays von Arrays und anderen komplexen Objekten zu arbeiten. Sprachen wie Q und ihre Verwandten K, J, APL sind für ihre Kürze bekannt. Häufig kann man ein Programm, das in einer herkömmlichen Sprache wie Java mehrere Bildschirme Codierung umfasst, in nur wenigen Zeilen auf ihnen schreiben. Genau das möchte ich in diesem Artikel demonstrieren.
Einführung
KDB+ ist eine spaltenorientierte Datenbank, die auf sehr große Datenmengen ausgerichtet ist, die in einer bestimmten Reihenfolge (hauptsächlich zeitbasiert) organisiert sind. Sie findet hauptsächlich Anwendung in Finanzinstitutionen – Banken, Investmentfonds, Versicherungen. Die Programmiersprache Q ist die interne Sprache von KDB+, die eine effiziente Arbeit mit diesen Daten ermöglicht. Die Ideologie von Q ist Kürze und Effizienz, wobei die Verständlichkeit opfert wird. Dies wird damit begründet, dass eine vektorielle Sprache in jedem Fall schwer zu erfassen ist, und die Kürze sowie die Dichte der Angaben erlauben es, einen viel größeren Teil des Programms auf einem Bildschirm gleichzeitig zu sehen, was letztendlich das Verständnis erleichtert.
In diesem Artikel werden wir ein vollständiges Programm in Q umsetzen, und möglicherweise möchten Sie es selbst ausprobieren. Dazu benötigen Sie Q. Die kostenlose 32-Bit-Version können Sie auf der Website des Unternehmens kx herunterladen – . Dort finden Sie auch, falls es Sie interessiert, hilfreiche Informationen zu Q, das Buch und diverse Artikel zu diesem Thema.
Problemstellung
Es gibt eine Quelle, die alle 25 Millisekunden eine Tabelle mit Daten sendet. Da KDB+ hauptsächlich in der Finanzwelt verwendet wird, nehmen wir an, dass es sich um eine Tabelle von Transaktionen handelt, die folgende Spalten hat: time (Zeit in Millisekunden), sym (Aktienbezeichnung an der Börse – IBM, AAPL,…), price (Preis, zu dem die Aktien gekauft wurden), size (Größe der Transaktion). Das Intervall von 25 Millisekunden wurde willkürlich gewählt, es ist weder zu klein noch zu groß. Dies bedeutet, dass die Daten bereits puffergeschichtet in den Service kommen. Man könnte die Pufferung auf der Serverseite leicht implementieren, einschließlich einer dynamischen, die von der aktuellen Last abhängt, aber der Einfachheit halber beschränken wir uns auf ein fixes Intervall.
Der Service sollte für jedes eingehende Symbol aus der Spalte sym stündlich eine Reihe aggregierter Funktionen berechnen – max price, avg price, sum size usw. nützliche Informationen. Der Einfachheit halber nehmen wir an, dass alle Funktionen inkrementell berechnet werden können, d.h. um einen neuen Wert zu erhalten, reicht es aus, zwei Zahlen zu kennen – den alten und den eingehenden Wert. Zum Beispiel besitzen die Funktionen max, average, sum diese Eigenschaft, während die Medianfunktion das nicht tut.
Wir gehen außerdem davon aus, dass der eingehende Datenstrom zeitlich geordnet ist. Dies ermöglicht es uns, nur mit der letzten Minute zu arbeiten. In der Praxis ist es jedoch ausreichend, mit der aktuellen und der vorherigen Minute zu arbeiten, falls es zu Verzögerungen bei Updates kommt. Zur Vereinfachung betrachten wir diesen Fall nicht.
Aggregationsfunktionen
Im Folgenden sind die erforderlichen Aggregationsfunktionen aufgeführt. Ich habe so viele wie möglich genommen, um die Belastung des Dienstes zu erhöhen:
- high – max price – maximaler Preis pro Minute.
- low – min price – minimaler Preis pro Minute.
- firstPrice – first price – erster Preis pro Minute.
- lastPrice – last price – letzter Preis pro Minute.
- firstSize – first size – erste Größe des Handels pro Minute.
- lastSize – last size – letzte Größe des Handels pro Minute.
- numTrades – count i – Anzahl der Handelsgeschäfte pro Minute.
- volume – sum size – Gesamtsumme der Handelsgrößen pro Minute.
- pvolume – sum price – Gesamtsumme der Preise pro Minute, nötig für avgPrice.
- turnover – sum price*size – Gesamtvolumen der Handelsgeschäfte pro Minute.
- avgPrice – pvolume%numTrades – Durchschnittspreis pro Minute.
- avgSize – volume%numTrades – Durchschnittsgröße des Handels pro Minute.
- vwap – turnover%volume – volumengewichteter Durchschnittspreis pro Minute.
- cumVolume – Gesamtes Volumen – akkumulierte Größe aller Transaktionen bis jetzt.
Lassen Sie uns sofort einen nicht offensichtlichen Punkt ansprechen – wie man diese Spalten beim ersten Mal und für jede nächste Minute initialisiert. Einige Spalten wie firstPrice müssen jedes Mal mit dem Wert null initialisiert werden, da ihr Wert nicht definiert ist. Andere wie volume sollten immer auf 0 gesetzt werden. Es gibt auch Spalten, die einen kombinierten Ansatz erfordern – zum Beispiel muss cumVolume aus der vorherigen Minute kopiert werden, während es für die erste Minute auf 0 gesetzt werden sollte. Wir werden all diese Parameter mit dem Datentyp Wörterbuch (analog zu einem Record) festlegen:
// 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 объяснен ниже
Ich habe sym und time zum Wörterbuch hinzugefügt, um die Benutzerfreundlichkeit zu erhöhen. Jetzt ist initWith eine fertige Zeile aus der finalen aggregierten Tabelle, in der nur noch die richtigen Werte für sym und time festgelegt werden müssen. Sie kann verwendet werden, um neue Zeilen in die Tabelle hinzuzufügen.
aggCols benötigen wir beim Erstellen der aggregierenden Funktion. Die Liste muss aufgrund der spezifischen Reihenfolge der Berechnungen in Q umgekehrt werden (von rechts nach links). Das Ziel ist es, die Berechnung in der Richtung von high zu cumVolume sicherzustellen, da einige Spalten von vorherigen abhängen.
Spalten, die in die neue Minute aus der vorherigen kopiert werden müssen, die Spalte sym wurde zur Vereinfachung hinzugefügt:
rollColumns:`sym`cumVolume;
Nun werden wir die Spalten in Gruppen aufteilen, je nachdem, wie sie aktualisiert werden sollen. Man kann drei Typen unterscheiden:
- Akkumulatoren (volume, turnover,..) – wir müssen den eingehenden Wert zum vorherigen addieren.
- Mit einem speziellen Punkt (high, low, ..) – der erste Wert in der Minute stammt aus den eingehenden Daten, die restlichen werden durch eine Funktion berechnet.
- Die restlichen. Werden immer durch eine Funktion berechnet.
Definieren wir Variablen für diese Klassen:
accumulatorCols:`numTrades`volume`pvolume`turnover;
specialCols:`high`low`firstPrice`firstSize;
Reihenfolge der Berechnungen
Wir werden die aggregierte Tabelle in zwei Schritten aktualisieren. Zur Effizienz werden wir zuerst die eingehende Tabelle komprimieren, sodass für jedes Symbol und jede Minute nur eine Zeile bleibt. Da alle unsere Funktionen inkrementell und assoziativ sind, wird das Ergebnis dieses zusätzlichen Schrittes nicht verändert. Die Tabelle könnte mit einem Select komprimiert werden:
select high:max price, low:min price … by sym,time.minute from table
Dieser Ansatz hat einen Nachteil: Die Menge der berechneten Spalten ist im Voraus festgelegt. Glücklicherweise ist die Q-Selektion auch als Funktion implementiert, in die dynamisch erstellte Argumente eingesetzt werden können:
?[table;whereClause;byClause;selectClause]
Ich werde das Format der Argumente nicht im Detail beschreiben; in unserem Fall sind nur die by- und select-Ausdrücke nicht trivial, und sie müssen Dictionaries der Form columns!expressions sein. So kann die zusammenfassende Funktion definiert werden:
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 ist die map-Funktion in Q für eine Liste
preprocess:?[;();`sym`time!`sym`time.minute;selExpression];
Zur Verdeutlichung habe ich die Funktion parse verwendet, die einen String mit einem Q-Ausdruck in einen Wert umwandelt, der an die eval-Funktion übergeben werden kann und im funktionalen Select benötigt wird. Zudem sei erwähnt, dass preprocess als Projektion (d.h. Funktion mit teilweise definierten Argumenten) der Select-Funktion angegeben ist, wobei ein Argument (Tabelle) fehlt. Wenn wir preprocess auf die Tabelle anwenden, erhalten wir eine komprimierte Tabelle.
Der zweite Schritt besteht darin, die aggregierte Tabelle zu aktualisieren. Zuerst schreiben wir den Algorithmus in Pseudocode:
für jedes sym in inputTable
idx: Zeilenindex in der aggregierten Tabelle für sym+currentTime;
aggTable[idx;`high]: aggTable[idx;`high] | inputTable[sym;`high];
aggTable[idx;`volume]: aggTable[idx;`volume] + inputTable[sym;`volume];
…
In Q ist es üblich, anstelle von Schleifen die Funktionen map/reduce zu verwenden. Da Q jedoch eine vektorielle Sprache ist und wir alle Operationen problemlos auf alle Symbole gleichzeitig anwenden können, können wir zunächst ganz auf eine Schleife verzichten und die Operationen auf alle Symbole gleichzeitig durchführen:
idx:calcIdx inputTable;
row:aggTable idx;
aggTable[idx;`high]: row[`high] | inputTable`high;
aggTable[idx;`volume]: row[`volume] + inputTable`volume;
…
Doch wir können noch weitergehen. In Q gibt es einen einzigartigen und äußerst mächtigen Operator – den Operator für verallgemeinerte Zuweisungen. Dieser ermöglicht es, einen Wertebereich in einer komplexen Datenstruktur mithilfe von Indexlisten, Funktionen und Argumenten zu ändern. In unserem Fall sieht er so aus:
idx:calcIdx inputTable;
rows:aggTable idx;
// .[target;(idx0;idx1;..);function;argument] ~ target[idx 0;idx 1;…]: function[target[idx 0;idx 1;…];argument], in unserem Fall ist die Funktion – das zugewiesene Element
.[aggTable;(idx;aggCols);:;flip (row[`high] | inputTable`high;row[`volume] + inputTable`volume;…)];
Leider benötigen wir für die Zuordnung in die Tabelle eine Liste von Zeilen und keine Spalten, und wir müssen die Matrix (von Spalten zu Zeilen) mit der Funktion flip transponieren. Dies ist bei großen Tabellen problematisch, deshalb verwenden wir stattdessen eine allgemeine Zuordnung für jede Spalte einzeln, indem wir die Funktion map anwenden (die wie ein Apostroph aussieht):
.[aggTable;;:;]'[(idx;)each aggCols; (row[`high] | inputTable`high;row[`volume] + inputTable`volume;…)];
Wir verwenden erneut die Projektion der Funktion. Beachten Sie auch, dass in Q die Erstellung einer Liste ebenfalls eine Funktion ist und wir sie mit der Funktion each(map) aufrufen können, um eine Liste von Listen zu erhalten.
Um sicherzustellen, dass die Menge der berechneten Spalten nicht festgelegt ist, erstellen wir den Ausdruck dynamisch. Zuerst definieren wir Funktionen zur Berechnung jeder Spalte unter Verwendung der Variablen row und inp, um auf die aggregierten und Eingabedaten zuzugreifen:
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");
Einige Spalten sind speziell, ihr erster Wert sollte nicht durch eine Funktion berechnet werden. Wir können feststellen, dass er der erste in der Spalte row[`numTrades] ist – wenn sie 0 enthält, ist der Wert der erste. In Q gibt es eine Auswahlfunktion — ?[Boolean list;list1;list2] – die je nach Bedingung im ersten Argument einen Wert aus Liste 1 oder 2 auswählt:
// high -> ?[isFirst;inp`high;row[`high]|inp`high]
// @ - тоже обобщенное присваивание для случая когда индекс неглубокий
@[`aggExpression;specialCols;{[x;y]"?[isFirst;inp`",y,";",x,"]"};string specialCols];
Hier habe ich eine allgemeine Zuweisung mit meiner Funktion (dem Ausdruck in geschweiften Klammern) aufgerufen. Sie erhält den aktuellen Wert (erstes Argument) und ein zusätzliches Argument, das ich im vierten Parameter übergebe.
Lassen Sie uns zusätzlich die Akkumulatorspalten hinzufügen, da für sie die Funktion dieselbe ist:
// volume -> row[`volume]+inp`volume
aggExpression[accumulatorCols]:{"row[`",x,"]+inp`",x } each string accumulatorCols;
Das ist eine ganz normale Zuweisung nach Q-Maßstäben, nur dass ich sofort eine Liste von Werten zuweisen. Schließlich erstellen wir die Hauptfunktion:
// ":",/: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),")]}";
Mit diesem Ausdruck erstelle ich dynamisch eine Funktion aus einer Zeichenkette, die den oben genannten Ausdruck enthält. Das Ergebnis sieht so aus:
{[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])]}
Die Reihenfolge der Berechnung der Spalten ist umgekehrt, da in Q die Reihenfolge der Berechnung von rechts nach links ist.
Jetzt haben wir zwei grundlegende Funktionen für die Berechnungen, es fehlt nur noch etwas Infrastruktur, und der Dienst ist bereit.
Letzte Schritte
Wir haben die Funktionen preprocess und updateAgg, die die ganze Arbeit erledigen. Allerdings müssen wir sicherstellen, dass der Übergang über die Minuten korrekt erfolgt und die Indizes für die Aggregation berechnet werden. Zunächst definieren wir die Funktion init:
init:{
tradeAgg:: 0#enlist[initWith]; // Erstellen einer leeren typisierten Tabelle, enlist wandelt das Dictionary in eine Tabelle um, und 0# bedeutet, dass 0 Elemente daraus entnommen werden
currTime::00:00; // Wir beginnen mit 0, :: bedeutet, dass es sich um eine Zuweisung der globalen Variable handelt
currSyms::`u#`symbol$(); // `u# wandelt die Liste in einen Baum um, um die Suche nach Elementen zu beschleunigen
offset::0; // Index in tradeAgg, an dem die aktuelle Minute beginnt
rollCache:: `sym xkey update `u#sym from rollColumns#tradeAgg; // Cache für die letzten Werte der Rollspalten, Tabelle mit dem Schlüssel sym
}
Lassen Sie uns auch die Funktion roll definieren, die die aktuelle Minute ändern wird:
roll:{[tm]
if[currTime>tm; :init[]]; // Wenn wir Mitternacht überschreiten, rufen wir einfach init auf
rollCache,::offset _ rollColumns#tradeAgg; // Aktualisieren des Caches – nehmen Sie Rollspalten aus aggTable, schneiden, und fügen Sie sie in rollCache ein
offset::count tradeAgg;
currSyms::`u#`$();
}
Wir benötigen eine Funktion, um neue Symbole hinzuzufügen:
addSyms:{[syms]
currSyms,::syms; // füge zur Liste der bekannten hinzu
// ergänze die Tabelle sym, time und rollColumns unter Verwendung der allgemeinen Zuweisung.
// Die Funktion ^ setzt Standardwerte für roll-Spalten, falls das Symbol nicht im Cache vorhanden ist. Der Wert flip table gibt die Liste der Spalten in der Tabelle zurück.
`tradeAgg upsert @[count[syms]#enlist initWith;`sym`time,cols rc;:;(syms;currTime), (initWith cols rc)^value flip rc:rollCache ([] sym: syms)];
}
Und schließlich die Funktion upd (traditioneller Name dieser Funktion für Q-Services), die vom Client aufgerufen wird, um Daten hinzuzufügen:
upd:{[tblName;data] // tblName benötigen wir nicht, aber in der Regel verarbeitet der Service mehrere Tabellen
tm:exec distinct time from data:() xkey preprocess data; // Daten vorverarbeiten & Zeit berechnen
updMinute[data] each tm; // füge Daten für jede Minute hinzu
};
updMinute:{[data;tm]
if[tmcurrTime; roll tm; currTime::tm]; // wechsle die Minute, wenn nötig
data:select from data where time=tm; // Filterung
if[count msyms:syms where not (syms:data`sym)in currSyms; addSyms msyms]; // neue Symbole
updateAgg[`tradeAgg;offset+currSyms?syms;data]; // aktualisiere die aggregierte Tabelle. Die Funktion ? sucht den Index der Elemente in der Liste rechts in der Liste links.
};
Das ist alles. Hier ist der vollständige Code unseres Services, wie versprochen, nur ein paar Zeilen:
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 Preis";"min Preis";"erster Preis";"letzter Preis";"erste Größe";"letzte Größe";"Anzahl i";"Summe Größe";"Summe Preis";"Summe Preis*Größe");
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];
};
Testen
Lassen Sie uns die Leistung des Dienstes überprüfen. Dazu starten wir ihn in einem separaten Prozess (speichern Sie den Code in der Datei service.q) und rufen die Funktion init auf:
q service.q –p 5566
q)init[]
Starten Sie in einer anderen Konsole einen zweiten Q-Prozess und verbinden Sie sich mit dem ersten:
h:hopen `:host:5566
h:hopen 5566 // wenn beide auf demselben Host sind
Zuerst erstellen wir eine Liste von Symbolen – 10.000 Stück – und fügen eine Funktion zur Erstellung einer zufälligen Tabelle hinzu. In der zweiten Konsole:
syms:`IBM`AAPL`GOOG,-9997?`8
rnd:{[n;t]([] sym:n?syms; time:t+asc n#til 25; price:n?10f; size:n?10)}
Ich habe der Liste drei echte Symbole hinzugefügt, um das Finden in der Tabelle zu erleichtern. Die Funktion rnd erstellt eine zufällige Tabelle mit n Zeilen, wobei sich die Zeit von t bis t+25 Millisekunden ändert.
Jetzt können wir versuchen, Daten an den Dienst zu senden (wir fügen die ersten zehn Stunden hinzu):
{h (`upd;`trade;rnd[10000;x])} each `time$00:00 + til 60*10
Wir können im Dienst überprüfen, ob die Tabelle aktualisiert wurde:
c 25 200
select from tradeAgg where sym=`AAPL
-20#select from tradeAgg where sym=`AAPL
Das Ergebnis:
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|2919Lassen Sie uns jetzt einen Lasttest durchführen, um herauszufinden, wie viele Daten der Dienst pro Minute verarbeiten kann. Ich erinnere daran, dass wir ein Update-Intervall von 25 Millisekunden festgelegt haben. Dementsprechend sollte der Dienst (im Durchschnitt) mindestens 20 Millisekunden für ein Update benötigen, um den Benutzern Zeit zu geben, Daten anzufordern. Geben Sie Folgendes im zweiten Prozess ein:
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 – das sind zwei Minuten. Man könnte zunächst für 1000 Zeilen alle 25 Millisekunden starten:
start 1000
In meinem Fall liegt das Ergebnis bei etwa ein paar Millisekunden pro Update. Daher werde ich die Anzahl der Zeilen sofort auf 10.000 erhöhen:
start 10000
Das Ergebnis:
min| 00:00:00.004
avg| 9.191458
med| 9f
max| 00:00:00.030
Wieder einmal nichts Besonderes, obwohl es sich um 24 Millionen Zeilen pro Minute handelt, 400.000 pro Sekunde. Über 25 Millisekunden wurde das Update nur 5 Mal verzögert, anscheinend beim Wechsel der Minute. Erhöhen wir auf 100.000:
start 100000
Das Ergebnis:
min| 00:00:00.013
avg| 25.11083
med| 24f
max| 00:00:00.108
q)sum times
00:02:00.532
Wie wir sehen können, hat der Service Schwierigkeiten, aber dennoch gelingt es ihm, sich über Wasser zu halten. Solch ein Datenvolumen (240 Millionen Zeilen pro Minute) ist extrem hoch; in solchen Fällen ist es üblich, mehrere Klone (oder sogar Dutzende von Klonen) des Services zu starten, wobei jeder nur einen Teil der Zeichen verarbeitet. Dennoch ist das Ergebnis beeindruckend für eine interpretierten Sprache, die in erster Linie auf Datenspeicherung ausgerichtet ist.
Es könnte die Frage aufkommen, warum die Zeit nicht linear mit der Größe jedes Updates zunimmt. Der Grund dafür ist, dass die komprimierende Funktion faktisch eine C-Funktion ist, die viel effizienter arbeitet als updateAgg. Ab einer bestimmten Update-Größe (ungefähr 10.000) erreicht updateAgg sein Limit, und die Ausführungszeit hängt nicht mehr von der Größe des Updates ab. Dank des vorausgehenden Schrittes Q kann der Dienst solche Datenmengen verarbeiten. Dies verdeutlicht, wie wichtig es ist, beim Umgang mit großen Datenmengen den richtigen Algorithmus auszuwählen. Ein weiterer Punkt ist die korrekte Speicherung von Daten im Speicher. Wenn die Daten nicht spaltenweise oder nicht zeitlich geordnet gespeichert würden, würden wir mit einem Phänomen namens TLB-Cache-Fehlzugriff konfrontiert – ein Fehlen der Speicheradressen im Adress-Cache des Prozessors. Die Adresssuche dauert im Falle eines Fehlschlags etwa 30 Mal länger und kann den Dienst bei verstreuten Daten erheblich verlangsamen.
Fazit
In diesem Artikel habe ich gezeigt, dass die KDB+ und Q-Datenbank nicht nur für die Speicherung großer Datenmengen und deren einfachen Zugriff über SELECT geeignet sind, sondern auch für die Erstellung von Datenverarbeitungsdiensten, die in der Lage sind, Hunderte Millionen Zeilen/Gigabyte an Daten selbst in einem einzelnen Q-Prozess zu verarbeiten. Die Q-Sprache ermöglicht es, Algorithmen, die mit Datenverarbeitung verbunden sind, äußerst prägnant und effizient aufgrund ihrer vektoriellen Natur, des integrierten SQL-Dialektinterpreter und des sehr gelungenen Satzes an Bibliotheksfunktionen umzusetzen.
Ich möchte anmerken, dass das oben Ausgeführte nur einen Teil der Möglichkeiten von Q darstellt; es besitzt auch andere einzigartige Eigenschaften. Zum Beispiel das äußerst einfache IPC-Protokoll, das die Grenze zwischen einzelnen Q-Prozessen verwischt und es ermöglicht, Hunderttausende dieser Prozesse zu einem gemeinsamen Netzwerk zusammenzuführen, das sich über Dutzende von Servern an verschiedenen Orten der Welt erstrecken kann.
Quelle: habr.com
