Vela → smarter Cache für Zeitreihen und mehr

Im Fintech-Bereich müssen wir oft ziemlich große Datenmengen zu Wechselkursen verarbeiten. Wir erhalten Daten aus verschiedenen Quellen, und jede von ihnen hat ihre eigene Auffassung darüber, wie Werte für die Kurse von morgen, übermorgen, nächsten Monat und sogar in den nächsten drei Jahren extrapoliert werden sollen. Wenn jemand in der Lage wäre, Kurse vorherzusagen, wäre es an der Zeit, das Geschäft zu schließen und einfach Geld hin und her zu tauschen. es wichtig,Einige Quellen genießen mehr Vertrauen, andere liefern nur Müll, seltene Einschübe mit fast richtigen Werten, aber dafür für exotische Paare. Unsere Aufgabe besteht darin, diese zigtausend Werte pro Sekunde zu filtern und zu bestimmen, was genau den Kunden gezeigt werden soll. Wir müssen den einzigen richtigen Wert aus Tonnen von Schmutz und Schlamm herausfiltern, wie das Flamingos beim Essen tun.

Ein charakteristisches Merkmal von Flamingos ist der massive, nach unten gebogene Schnabel, mit dem sie Nahrung aus Wasser oder Schlamm filtern.
 — Wiki

So entstand die Bibliothek Vela, die einen Cache von Zuständen für mehrere Werte in vorgegebenen Zeitintervallen speichert. Im Hintergrund filtert sie in Echtzeit schlechte und veraltete Daten heraus und bietet Zugang zu den letzten N gültigen Werten für jeden Schlüssel (Währungspaare, in unserem Fall).

Angenommen, wir sammeln Kurse für drei Währungspaare. Die einfachste Definition Vela zur Speicherung des aktuellen Status würde etwa so aussehen:

defmodule Pairs do
  use Vela,
    eurusd: [sorter: &Kernel. 0
end

Aktualisierung der Werte

Vela.put/3 Die Funktion wird der Reihe nach Folgendes tun:

• sie wird den Validator auf dem Wert aufrufen, wenn er definiert ist (siehe Kapitel Validierung unten);
• sie fügt den Wert entweder in die Reihe der guten Werte hinzu, wenn die Validierung erfolgreich war, oder in die Hilfsreihe :__errors__ im anderen Fall;
• sie wird die Sortierung aufrufen, wenn ein Sorter für diesen Schlüssel definiert ist, oder legt den Wert einfach an den Kopf der Liste (LIFO, siehe Kapitel Sortierung unten);
• schneidet die Reihe gemäß dem Parameter :limit ab, der bei der Erstellung übergeben wurde;
• sie gibt die aktualisierte Struktur zurück. Vela.

iex|1 > pairs = %Pairs{}
iex|2 > Vela.put(pairs, :eurcad, 1.0)
#⇒ %Pairs{..., eurcad: [1.0], ...}
iex|3 > Vela.put(pairs, :eurcad, -1.0)
#⇒ %Pairs{__errors__: [eurcad: -1.0], ...}
iex|4 > pairs |> Vela.put(:eurusd, 2.0) |> Vela.put(:eurusd, 1.0)
#⇒ %Pairs{... eurusd: [1.0, 2.0]}

Außerdem Vela implementiert Zugriff, sodass jede der Standardfunktionen für die tiefe Aktualisierung von Strukturen genutzt werden kann Kernel: Kernel.get_in/2, Kernel.put_in/3, Kernel.update_in/3, Kernel.pop_in/2, und Kernel.get_and_update_in/3.

Validierung

Der Validator kann definiert werden als:

• eine externe Funktion mit einem Argument (&MyMod.my_fun/1), sie erhält nur den Wert zur Validierung;
• eine externe Funktion mit zwei Argumenten, &MyMod.my_fun/2, sie erhält das Paar serie, value zur Validierung;
• ein Modul, das implementiert Vela.Validator;
• Konfigurationsparameter Schwellenwert, und – optional – compare_by, siehe Kapitel Vergleich darunter.

Wenn die Validierung erfolgreich war, wird der Wert unter dem entsprechenden Schlüssel zur Liste hinzugefügt, andernfalls wird das Tupel {serie, value} an :__errors_.

Vergleich

Werte, die in diesen Reihen gespeichert werden, können beliebig sein. Um zu lernen, Vela sie zu vergleichen, muss compare_by ein Parameter in die Definition der Reihe übergeben werden (es sei denn, die Werte können mit dem Standard Kernel.</2); dieser Parameter sollte den Typ haben (Vela.value() -> number()). Standardmäßig ist dies einfach & &1.

Zusätzlich kann zur Definition der Reihe ein Parameter comparator für die Berechnung der Delta-Werte übergeben werden (min/max); beispielsweise kann man durch das Übergeben von Date.diff/2 als Comparator korrekte Deltas für Daten erhalten.

Eine weitere bequeme Möglichkeit ist die Übergabe eines Parameters, Schwellenwertder das maximal zulässige Verhältnis des neuen Wertes zu {min, max} intervall definiert. Da dies in Prozent angegeben ist, nutzt die Überprüfung nicht comparator, verwendet aber immer noch compare_by. Beispielsweise muss, um einen Schwellenwert für Datumswerte anzugeben, compare_by: &DateTime.to_unix/1 (um einen ganzzahligen Wert zu erhalten) und threshold: 1, damit neue Werte nur erlaubt sind, wenn sie sich im ±band Intervall der aktuellen Werte befinden.

Schließlich kann man Vela.equal?/2 zum Vergleichen zweier Caches verwenden. Wenn die Werte die Funktion equal?/2 oder compare/2, dann werden diese Funktionen für den Vergleich verwendet, andernfalls verwenden wir einfach ==/2.

Werte abrufen

Die Verarbeitung des aktuellen Zustands beginnt normalerweise mit dem Aufruf Vela.purge/1, der veraltete Werte entfernt (wenn Validator mit timestamps). Dann kann man Vela.slice/1, die zurückgibt keyword mit den Namen der Reihen als Schlüsseln und den ersten, aktuellen Werten.

Es kann auch genutzt werden get_in/2/pop_in/2 für den Zugriff auf Werte in jeder Reihe auf niedriger Ebene.

Anwendung

Vela kann sich als äußerst nützlich als Cache für Zeitreihen im Zustand eines Prozess-Typs erweisen GenServer/Agent. Wir möchten niemals veraltete Kurswerte verwenden, und dazu halten wir einfach einen Prozess mit einem Zustand, der von Vela, mit dem Validator, der unten gezeigt wird, verarbeitet wird.

@impl Vela.Validator
def valid?(_key, %Rate{} = rate),
  do: Rate.age(rate) < @death_age

und Vela.purge/1 der alle veralteten Werte jedes Mal entfernt, wenn wir Daten benötigen. Für den Zugriff auf die aktuellen Werte rufen wir einfach Vela.slice/1, und wenn eine kleine Historie des Kurses benötigt wird (die gesamte Reihe), geben wir sie einfach zurück – bereits sortiert – mit validierten Werten.

• GitHub
• Hexdocs

Viel Erfolg beim Caching von Zeitreihen!

Quelle: habr.com