Über die KDB+-Datenbank, die Programmiersprache Q, sowie deren Stärken und Schwächen kann man in meinem vorherigen Artikel lesen und kurz in der Einleitung. In diesem Artikel werden wir einen Dienst auf Q implementieren, der den eingehenden Datenstrom verarbeitet und verschiedene aggregierende Funktionen minutengenau in „Echtzeit“ berechnet (d.h. er wird es schaffen, alles vor der nächsten Datencharge zu berechnen). Die Hauptbesonderheit von Q besteht darin, dass es sich um eine vektororientierte Sprache handelt, die es ermöglicht, nicht mit einzelnen Objekten, sondern mit ihren Arrays, Arrays von Arrays und anderen komplexen Objekten zu arbeiten. Sprachen wie Q und seine Verwandten K, J, APL sind für ihre Kürze berühmt. Häufig kann ein Programm, das in einer herkömmlichen Sprache wie Java mehrere Bildschirme Code benötigt, in diesen Sprachen in wenigen Zeilen geschrieben werden. 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 auf bestimmte Weise (hauptsächlich nach Zeit) geordnet sind. Sie wird hauptsächlich in Finanzinstitutionen - Banken, Investmentfonds, Versicherungsunternehmen - verwendet. Die Sprache Q ist die interne Sprache von KDB+, die es ermöglicht, effizient mit diesen Daten zu arbeiten. Die Ideologie von Q ist Kürze und Effizienz, wobei Klarheit dafür in gewissem Maße geopfert wird. Dies wird damit begründet, dass eine vektororientierte Sprache in jedem Fall schwer zu verstehen sein wird und die Kürze und Dichte der Notation es ermöglicht, einen viel größeren Teil des Programms auf einem Bildschirm 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 Sie interessiert sind, Informationen zur Referenz zu Q, das Buch und verschiedene Artikel zu diesem Thema.
Aufgabenstellung
Es gibt eine Quelle, die alle 25 Millisekunden eine Tabelle mit Daten sendet. Da KDB+ hauptsächlich im Finanzwesen eingesetzt wird, nehmen wir an, dass es sich um eine Handelstabelle (trades) handelt, in der folgende Spalten enthalten sind: time (Zeit in Millisekunden), sym (Aktienbezeichnung an der Börse - IBM, AAPL,…), Preis (der Preis, zu dem die Aktien gekauft wurden), Größe (Größe des Geschäfts). Das Intervall von 25 Millisekunden wurde willkürlich gewählt, es ist weder zu klein noch zu groß. Seine Existenz bedeutet, dass die Daten bereits im Dienst gepuffert ankommen. Man könnte die Pufferung auf der Seite des Dienstes leicht implementieren, auch dynamisch, abhängig von der aktuellen Last, aber zur Vereinfachung bleiben wir bei einem festen Intervall.
Der Dienst muss für jedes eingehende Zeichen aus der Spalte sym minutengenau eine Reihe von Aggregatfunktionen berechnen – max Preis, avg Preis, sum Größe usw. nützliche Informationen. Zur Vereinfachung gehen wir davon aus, dass alle Funktionen inkrementell berechnet werden können, d.h. um einen neuen Wert zu erhalten, genügt es, zwei Zahlen zu kennen – den alten und den eingehenden Wert. Zum Beispiel haben die Funktionen max, average, sum diese Eigenschaft, die Mediane-Funktion nicht.
Wir nehmen auch an, dass der eingehende Datenstrom nach der Zeit geordnet ist. Dies ermöglicht es uns, nur mit der letzten Minute zu arbeiten. In der Praxis genügt es, mit der aktuellen und der vorherigen Minute zu arbeiten, falls einige Aktualisierungen verspätet sind. Zur Vereinfachung werden wir diesen Fall nicht berücksichtigen.
Aggregatfunktionen
Im Folgenden sind die benötigten Aggregatfunktionen aufgeführt. Ich habe so viele wie möglich ausgewählt, um die Belastung für den Dienst zu erhöhen:
- high – max Preis – maximaler Preis pro Minute.
- low – min Preis – minimaler Preis pro Minute.
- firstPrice – erster Preis – erster Preis pro Minute.
- lastPrice – letzter Preis – letzter Preis pro Minute.
- firstSize – erste Größe – erste Größe des Geschäfts pro Minute.
- lastSize – letzte Größe – letzte Größe des Geschäfts pro Minute.
- numTrades – Anzahl i – Anzahl der Geschäfte pro Minute.
- volume – sum Größe – Summe der Größen der Geschäfte pro Minute.
- pvolume – sum Preis – Summe der Preise pro Minute, benötigt für avgPrice.
- turnover – sum Preis*Größe – Gesamtvolumen der Geschäfte pro Minute.
- avgPrice – pvolume%numTrades – Durchschnittspreis pro Minute.
- avgSize – volume%numTrades – Durchschnittsgröße des Geschäfts pro Minute.
- vwap – turnover%volume – gewogene durchschnittliche Preis pro Minute.
- cumVolume – sum volume – akkumulierte Größe der Geschäfte über alle Zeiten.
Lassen Sie uns sofort einen nicht offensichtlichen Punkt erörtern – wie man diese Spalten beim ersten Mal und für jede folgende Minute initialisiert. Einige Spalten wie firstPrice müssen jedes Mal mit dem Wert null initialisiert werden, ihr Wert ist nicht definiert. Andere wie volume müssen immer auf 0 gesetzt werden. Es gibt auch Spalten, die einen kombinierten Ansatz erfordern – beispielsweise muss cumVolume aus der vorherigen Minute kopiert werden, während es für die erste auf 0 gesetzt werden sollte. Lassen Sie uns all diese Parameter unter Verwendung des Datentyps Dictionary (analog einer Aufzeichnung) 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 in das Dictionary zur besseren Übersichtlichkeit aufgenommen, jetzt ist initWith die fertige Zeile aus der endgültigen aggregierten Tabelle, in der nur noch die richtigen Werte für sym und time festgelegt werden müssen. Sie kann für das Hinzufügen neuer Zeilen zur Tabelle verwendet werden.
aggCols werden wir bei der Erstellung der Aggregatfunktion benötigen. Die Liste muss aufgrund der Besonderheiten der Rechenreihenfolge der Ausdrücke in Q (von rechts nach links) umgekehrt werden. Ziel ist es, die Berechnung von high nach cumVolume zu gewährleisten, da einige Spalten von vorherigen abhängen.
Die Spalten, die in die neue Minute aus der vorherigen kopiert werden müssen, sind für die Übersichtlichkeit die Spalte sym hinzugefügt worden:
rollColumns:`sym`cumVolume;
Jetzt teilen wir die Spalten in Gruppen auf, je nachdem, wie sie aktualisiert werden sollen. Man kann drei Typen unterscheiden:
- Akkumulatoren (volume, turnover, ..) – wir müssen den eingehenden Wert zum vorhergehenden addieren.
- Mit besonderem Punkt (high, low, ..) – der erste Wert in der Minute wird aus den eingehenden Daten entnommen, die anderen werden mithilfe der Funktion berechnet.
- Die übrigen werden immer mithilfe der Funktion berechnet.
Lassen Sie uns Variablen für diese Klassen definieren:
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 zunächst die eingehende Tabelle so komprimieren, dass für jedes Zeichen und jede Minute nur eine Zeile übrig bleibt. Dass alle unsere Funktionen inkrementell und assoziativ sind, garantiert uns, dass sich das Ergebnis durch diesen zusätzlichen Schritt nicht ändert. Die Tabelle könnte mithilfe von Select komprimiert werden:
select high:max price, low:min price … by sym,time.minute from table
Dieser Ansatz hat einen Nachteil – das Set an berechneten Spalten ist im Voraus definiert. Glücklicherweise wird in Q Select 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 sollten Wörterbücher des Typs columns!expressions sein. Auf diese Weise kann die komprimierende Funktion wie folgt 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 in die Funktion eval übergeben werden kann und der in einem funktionalen Select erforderlich ist. Außerdem sei angemerkt, dass preprocess als Projektion (d.h. Funktion mit teilweise festgelegten Argumenten) für die select-Funktion definiert ist, 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. Lassen Sie uns zunächst den Algorithmus im Pseudocode schreiben:
for each 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 map/reduce-Funktionen zu verwenden. Da Q jedoch eine vektorielle Sprache ist und wir alle Operationen problemlos auf alle Symbole gleichzeitig anwenden können, können wir in der ersten Annäherung ganz ohne Schleife auskommen und alle Operationen gleichzeitig mit allen Symbolen durchführen:
idx:calcIdx inputTable;
row:aggTable idx;
aggTable[idx;`high]: row[`high] | inputTable`high;
aggTable[idx;`volume]: row[`volume] + inputTable`volume;
…
Wir können noch weiter gehen; in Q gibt es einen einzigartigen und äußerst leistungsfähigen Operator – den Operator der allgemeinen Zuweisung. Er ermöglicht es, eine Menge von Werten in einer komplexen Datenstruktur zu ändern, indem eine Liste von Indizes, Funktionen und Argumenten verwendet wird. 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 die Zuweisung
.[aggTable;(idx;aggCols);:;flip (row[`high] | inputTable`high;row[`volume] + inputTable`volume;…)];
Leider benötigt die Zuweisung in einer Tabelle eine Liste von Zeilen und keine Spalten, daher müssen wir die Matrix (Liste von Spalten in eine Liste von Zeilen) mit der Funktion flip transponieren. Für eine große Tabelle ist dies aufwändig, daher wenden wir stattdessen die allgemeine Zuweisung für jede Spalte einzeln an und verwenden die Funktion map (die wie ein Apostroph aussieht):
.[aggTable;;:;]'[(idx;)jeder aggCols; (row[`high] | inputTable`high;row[`volume] + inputTable`volume;…)];
Wir verwenden erneut die Funktionsprojektion. Beachten Sie auch, dass die Erstellung einer Liste in Q ebenfalls eine Funktion ist, die wir mit der Funktion each(map) aufrufen können, um eine Liste von Listen zu erhalten.
Um zu verhindern, dass die Gruppe der berechneten Spalten festgelegt ist, erstellen wir den Ausdruck dynamisch oben. Zuerst definieren wir Funktionen zur Berechnung jeder Spalte und verwenden die Variablen row und inp, um auf aggregierte 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 besonders, ihr erster Wert sollte nicht durch eine Funktion berechnet werden. Wir können feststellen, dass es sich um den ersten Wert in der Spalte row[`numTrades] handelt – wenn dieser 0 ist, ist der Wert der erste. In Q gibt es eine Auswahlfunktion – ?[Boolean list;list1;list2] – die einen Wert aus Liste 1 oder 2 basierend auf der Bedingung im ersten Argument 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 aufgerufen (der Ausdruck in geschweiften Klammern). Dabei wird der aktuelle Wert (erster Parameter) und ein zusätzlicher Parameter übergeben, den ich im vierten Parameter angebe.
Zusätzlich fügen wir Akkumulator-Spalten hinzu, 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 gewöhnliche Zuweisung in Q, nur dass ich sofort eine Liste von Werten zuweise. 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 Zeile, die den Ausdruck enthält, den ich oben angegeben habe. Das Ergebnis sieht folgendermaßen aus:
{[aggTable;idx;inp] rows:aggTable idx; isFirst_0=row`numTrades; .[aggTable;;:;]'[(idx;)jeder 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 Berechnung von rechts nach links erfolgt.
Jetzt haben wir zwei Hauptfunktionen, die für die Berechnungen erforderlich sind, und es bleibt nur noch, ein wenig Infrastruktur hinzuzufügen, und der Dienst ist bereit.
Letzte Schritte
Wir haben die Funktionen preprocess und updateAgg, die die gesamte Arbeit erledigen. Es ist jedoch auch notwendig, den korrekten Übergang über die Minuten sicherzustellen und die Indizes für die Aggregation zu berechnen. Zuerst definieren wir die Funktion init:
init:{
tradeAgg:: 0#enlist[initWith]; // wir erstellen eine leere typisierte Tabelle, enlist verwandelt das Wörterbuch in eine Tabelle, und 0# bedeutet, 0 Elemente daraus zu übernehmen
currTime::00:00; // beginnen wir mit 0, :: bedeutet, dass die Zuweisung in eine globale Variable erfolgt
currSyms::`u#`symbol$(); // `u# - verwandelt die Liste in einen Baum, um die Suche nach Elementen zu beschleunigen
offset::0; // Index in tradeAgg, wo 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
}
Definieren wir auch die Funktion roll, die die aktuelle Minute ändern wird:
roll:{[tm]
if[currTime>tm; :init[]]; // wenn wir nach Mitternacht sind, rufen wir einfach init auf
rollCache,::offset _ rollColumns#tradeAgg; // aktualisieren wir den Cache – die Rollspalten aus aggTable übernehmen, zuschneiden, in rollCache einfügen
offset::count tradeAgg;
currSyms::`u#`$();
}
Wir brauchen eine Funktion zum Hinzufügen neuer Symbole:
addSyms:{[syms]
currSyms,::syms; // fügen wir der Liste der bekannten hinzu
// fügen wir die Tabelle sym, time und rollColumns mit genereller Zuweisung hinzu.
// Die Funktion ^ fügt Standardwerte für Rollspalten hinzu, wenn das Symbol nicht im Cache ist. value 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-Dienste), die vom Client aufgerufen wird, um Daten hinzuzufügen:
upd:{[tblName;data] // tblName brauchen wir nicht, aber normalerweise verarbeitet der Dienst mehrere Tabellen
tm:exec distinct time from data:() xkey preprocess data; // preprocess & calc time
updMinute[data] each tm; // fügen wir Daten für jede Minute hinzu
};
updMinute:{[data;tm]
if[tmcurrTime; roll tm; currTime::tm]; // ändern wir die Minute, falls erforderlich
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]; // aktualisieren wir die aggregierte Tabelle. Die Funktion ? sucht nach dem Index von Elementen in der rechten Liste in der linken Liste.
};
Das ist alles. Hier ist der gesamte Code unseres Dienstes, 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 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];
};
Tests
Lassen Sie uns die Leistung des Dienstes überprüfen. Dazu starten wir ihn in einem separaten Prozess (setzen Sie den Code in die Datei service.q) und rufen die Funktion init auf:
q service.q –p 5566
q)init[]
Starten Sie in einer zweiten 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 – 10000 Stück und fügen eine Funktion hinzu, um eine zufällige Tabelle zu erstellen. 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 von Symbolen drei echte hinzugefügt, um sie in der Tabelle leichter zu finden. Die Funktion rnd erzeugt eine zufällige Tabelle mit n Zeilen, wobei die Zeit von t bis t+25 Millisekunden variiert.
Nun können wir versuchen, Daten an den Dienst zu senden (fügen wir die ersten zehn Stunden hinzu):
{h (`upd;`trade;rnd[10000;x])} each `time$00:00 + til 60*10
Man kann im Dienst überprüfen, dass die Tabelle aktualisiert wurde:
c 25 200
select from tradeAgg where sym=`AAPL
-20#select from tradeAgg where sym=`AAPL
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 das 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. Es kann sich lohnen, zuerst für 1000 Zeilen alle 25 Millisekunden zu starten:
start 1000
In meinem Fall liegt das Ergebnis bei ein paar Millisekunden pro Update. Daher werde ich die Anzahl der Zeilen sofort auf 10.000 erhöhen:
start 10000
Ergebnis:
min| 00:00:00.004
avg| 9.191458
med| 9f
max| 00:00:00.030
Wieder nichts Besonderes, und das sind immerhin 24 Millionen Zeilen pro Minute, 400.000 pro Sekunde. Über 25 Millisekunden wurde das Update nur 5 Mal verzögert, offensichtlich beim Wechsel der Minute. Lassen Sie uns auf 100.000 erhöhen:
start 100000
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, hat der Dienst Mühe, aber dennoch gelingt es ihm, über Wasser zu bleiben. Solch ein Datenvolumen (240 Millionen Zeilen pro Minute) ist außergewöhnlich hoch, in solchen Fällen ist es üblich, mehrere Klone (oder sogar Dutzende von Klonen) des Dienstes zu starten, von denen jeder nur einen Teil der Symbole verarbeitet. Dennoch ist das Ergebnis beeindruckend für eine interpretierte Sprache, die in erster Linie auf die Speicherung von Daten ausgerichtet ist.
Es könnte die Frage aufkommen, warum die Zeit nichtlinear mit der Größe jedes Updates wächst. Der Grund dafür ist, dass die komprimierende Funktion im Grunde eine C-Funktion ist, die viel effizienter als updateAgg arbeitet. Ab einer bestimmten Größe des Updates (im Bereich von 10.000) erreicht updateAgg seine Obergrenze, und die Ausführungszeit hängt dann nicht mehr von der Größe des Updates ab. Gerade durch den vorgelagerten Schritt Q ist der Dienst in der Lage, solche Datenmengen zu verarbeiten. Das unterstreicht, wie wichtig es ist, bei der Arbeit mit großen Daten die richtige Algorithmuswahl zu treffen. Ein weiterer Punkt ist die korrekte Speicherung der Daten im Speicher. Wenn die Daten nicht spaltenweise oder nicht zeitlich geordnet gespeichert wären, würden wir es mit einem Phänomen namens TLB-Cache-Miss zu tun bekommen – dem Fehlen einer Speicherseitenadresse im Adress-Cache des Prozessors. Die Adresssuche dauert im Falle eines Misserfolgs etwa 30 Mal länger und kann den Dienst bei verstreuten Daten um ein Vielfaches verlangsamen.
Fazit
In diesem Artikel habe ich gezeigt, dass die KDB+-Datenbank und Q nicht nur zur Speicherung großer Datenmengen und zum einfachen Zugriff über selektive Abfragen geeignet sind, sondern auch zur Erstellung von Datenverarbeitungsdiensten, die in der Lage sind, Hunderte Millionen Zeilen/Gigabyte an Daten sogar in einem einzelnen Q-Prozess zu verarbeiten. Die Sprache Q ermöglicht es, Algorithmen zur Datenverarbeitung äußerst prägnant und effizient aufgrund ihrer vektoriellen Natur, des integrierten SQL-Dialektinterpreter und eines äußerst gelungenen Sets von Bibliotheksfunktionen zu implementieren.
Ich möchte anmerken, dass das oben Genannte nur einen Teil der Möglichkeiten von Q darstellt; es gibt auch andere einzigartige Eigenschaften. Zum Beispiel das extrem einfache IPC-Protokoll, das die Grenze zwischen einzelnen Q-Prozessen verwischt und es ermöglicht, Hunderte dieser Prozesse zu einem einzigen Netzwerk zusammenzuführen, das sich auf Dutzenden von Servern an verschiedenen Orten der Welt befinden kann.
Quelle: habr.com
