ProHoster > Blog > Administración > Vela → caché intelixente para series temporais e moito máis

Vela → caché intelixente para series temporais e moito máis

En fintech, moitas veces temos que procesar volumes bastante masivos de datos de tipos de cambio de moeda. Recibimos datos de diferentes fontes, e cada unha delas ten a súa propia idea de como extrapolar os tipos de cambio para mañá, pasadomañá, o próximo mes e mesmo os próximos tres anos. Se alguén puidese predecir as taxas correctamente, sería hora de pechar o negocio e simplemente cambiar o diñeiro de un lado a outro. Algunhas fontes son máis fiables, algunhas proporcionan lixo completo, con inclusións raras de valores case correctos, pero para parellas exóticas. O noso traballo é examinar estas decenas de miles de valores por segundo e determinar o que mostrar exactamente aos clientes. Necesitamos filtrar o valor correcto de toneladas de sucidade e limo, como fan os flamencos no xantar.

Vela → caché intelixente para series temporais e moito máis

Un trazo distintivo especial dos flamencos é o seu enorme peteiro curvado cara abaixo, co que filtran os alimentos da auga ou do barro.
 - Wiki

Así naceu a biblioteca Vela, que almacena unha caché de estado para varios valores a intervalos de tempo especificados. Baixo o capó, filtra os datos malos e obsoletos sobre a marcha e tamén proporciona acceso ás últimas N valores validados para cada clave (pares de moedas, no noso caso).

Digamos que recollemos taxas de tres pares de divisas. Definición máis sinxela Vela para almacenar o estado actual terá un aspecto así: 

defmodule Pairs do
  use Vela,
    eurusd: [sorter: &Kernel.<=/2],
    eurgbp: [limit: 3, errors: 1],
    eurcad: [validator: Pairs]

  @behaviour Vela.Validator

  @impl Vela.Validator
  def valid?(:eurcad, rate), do: rate > 0
end

Actualización de valores

Vela.put/3 A función fará o seguinte en secuencia:

  • provocará validator sobre o valor, se se define un (ver capítulo Validación abaixo);
  • engadirá o valor á fila de bos valores se a validación foi exitosa ou á fila de servizo :__errors__ en caso contrario;
  • provocará a clasificación se sorter definido para unha chave determinada, ou simplemente poñerá o valor na cabeceira da lista (LIFO, ver capítulo Ordenación abaixo);
  • recortará a fila segundo o parámetro :limit pasado sobre a creación;
  • devolverá a estrutura actualizada 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]}

Tamén Vela apeiros Access, polo que podes usar calquera das funcións estándar para a actualización profunda das estruturas do arsenal para actualizar os valores Kernel: Kernel.get_in/2, Kernel.put_in/3, Kernel.update_in/3, Kernel.pop_in/2e Kernel.get_and_update_in/3.

Validación

Un validador pódese definir como:

  • función externa cun argumento (&MyMod.my_fun/1), só recibirá o valor para validación;
  • función externa con dous argumentos, &MyMod.my_fun/2, ela conseguirá un par serie, value para validación;
  • implementación de módulos Vela.Validator;
  • parámetro de configuración thresholde - opcionalmente - compare_by, ver capítulo comparación a continuación.

Se a validación ten éxito, o valor engádese á lista baixo a chave correspondente; se non, a tupla {serie, value} vai a :__errors_.

Comparación

Os valores almacenados nestas filas poden ser calquera cousa. Ensinar Vela para comparalos, é necesario transferir compare_by parámetro na definición da serie (a non ser que os valores non se poidan comparar co estándar Kernel.</2); este parámetro debe ser de tipo (Vela.value() -> number()). Por defecto é sinxelo & &1.

Ademais, pode pasar un parámetro á definición da fila comparator para calcular valores delta (min/max); por exemplo, transmitindo Date.diff/2 como comparador, podes obter os deltas correctos para as datas.

Outra forma conveniente de traballar é pasar un parámetro threshold, que define a relación máxima permitida do novo valor a {min, max} intervalo. Dado que se especifica como unha porcentaxe, o cheque non se utiliza comparatorpero aínda usa compare_by. Por exemplo, para especificar un valor límite para datas e horas, debes especificalo compare_by: &DateTime.to_unix/1 (para obter un valor enteiro) e threshold: 1, o que fai que só se permitan novos valores se están en ±band intervalo dos valores actuais.

Finalmente, podes usar Vela.equal?/2 para comparar dous cachés. Se os valores definen unha función equal?/2 ou compare/2, entón estas funcións serán usadas para a comparación, se non, usamos estúpidamente ==/2.

Obtención de valores

O procesamento do estado actual comeza normalmente coa chamada Vela.purge/1, que elimina os valores obsoletos (se validator atado a timestamps). Despois podes chamar Vela.slice/1que volverá keyword cos nomes das filas como claves e os primeiros valores reais.

Tamén podes usar get_in/2/pop_in/2 para acceso de baixo nivel aos valores de cada fila.

App

Vela pode ser moi útil como caché de series temporais nun estado de proceso como GenServer/Agent. Queremos nunca usar valores obsoletos do curso, e para iso simplemente mantemos o proceso co estado procesado Vela, co validador que se mostra a continuación.

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

и Vela.purge/1 elimina silenciosamente todos os valores obsoletos cada vez que necesitamos os datos. Para acceder aos valores reais simplemente chamamos Vela.slice/1, e cando se require un pequeno historial do curso (toda a serie), simplemente o devolvemos -xa ordenado- cos valores validados.

Feliz almacenamento en caché de series temporais!

Fonte: www.habr.com

Engadir un comentario