Implementar agregados definidos pelo utilizador em JavaScript no Azure Stream Analytics

O Azure Stream Analytics suporta agregados definidos pelo utilizador (UDA) escritos em JavaScript para que possa implementar lógica de negócio complexa com estado. Com uma UDA, tem controlo total sobre a estrutura de dados de estado, acumulação de estados, desacumulação de estados e cálculo agregado de resultados.

Usa um UDA em JavaScript quando as funções agregadas incorporadas não satisfazem as tuas necessidades e quiseres agregar eventos em janelas com o teu próprio algoritmo.

Este artigo mostra-lhe como criar uma UDA e como a chamar com operações baseadas em janelas numa consulta de Análise de Fluxos.

Pré-requisitos

Antes de começar, certifique-se de que tem:

Escolha um tipo de agregado definido pelo utilizador em JavaScript

Um agregado definido pelo utilizador é executado com base numa especificação de janela temporal para agregar os eventos dessa janela e produzir um único valor de resultado. A Análise de Fluxos suporta dois tipos de interfaces UDA: AccumulateOnly e AccumulateDeaccumulate. Ambos os tipos suportam janelas de tumbling, hopping, deslizantes e de sessão. Escolha o tipo com base no algoritmo que utiliza.

Os agregados AccumulateDeaccumulate apresentam melhor desempenho do que os agregados AccumulateOnly quando são utilizados com janelas de salto, janelas deslizantes e janelas de sessão, porque o Stream Analytics pode remover eventos do estado em vez de o recalcular.

Acumular apenas agregados

Os agregados AccumulateOnly só podem acumular novos eventos no seu estado. O algoritmo não permite a desacumulação de valores. Escolha este tipo quando não puder remover as informações de um evento do valor de estado. O seguinte código é o modelo JavaScript para agregados AccumulateOnly:

// Sample UDA which state can only be accumulated.
function main() {
    this.init = function () {
        this.state = 0;
    }

    this.accumulate = function (value, timestamp) {
        this.state += value;
    }

    this.computeResult = function () {
        return this.state;
    }
}

AcumularDesacumular agrega

AcumularDesacumular agregados desacumulam um valor previamente acumulado do estado. Por exemplo, pode remover um par chave-valor de uma lista de valores de evento ou subtrair um valor de um agregado de soma. O código seguinte é o modelo em JavaScript dos agregados AccumulateDeaccumulate:

// Sample UDA which state can be accumulated and deaccumulated.
function main() {
    this.init = function () {
        this.state = 0;
    }

    this.accumulate = function (value, timestamp) {
        this.state += value;
    }

    this.deaccumulate = function (value, timestamp) {
        this.state -= value;
    }

    this.deaccumulateState = function (otherState){
        this.state -= otherState.state;
    }

    this.computeResult = function () {
        return this.state;
    }
}

Compreender a declaração de função JavaScript

Uma declaração de objeto Function define cada UDA em JavaScript. A lista seguinte descreve os principais elementos numa definição da UDA.

Alias de função

O alias da função é o identificador UDA. Quando chamar uma UDA numa consulta do Stream Analytics, use sempre o pseudónimo com o prefixo uda..

Tipo de função

Para um UDA, defina o tipo de função para JavaScript UDA.

Tipo de saída

Define o tipo de saída para um tipo específico que o trabalho de Stream Analytics suporta, ou para Qualquer se quiseres tratar o tipo na tua consulta.

Nome da função

O nome do objeto Function. O nome da função deve corresponder ao alias UDA.

Método: init()

O init() método inicializa o estado do agregado. O Stream Analytics chama este método quando a janela começa.

Método: accumulate()

O accumulate() método calcula o estado UDA com base no estado anterior e nos valores atuais do evento. O Stream Analytics chama este método quando um evento entra numa janela temporal (TumblingWindow, HoppingWindow, SlidingWindow, ou SessionWindow).

Método: desacumular()

O deaccumulate() método recalcula o estado com base no estado anterior e nos valores atuais do evento. Stream Analytics invoca este método quando um evento sai de um SlidingWindow ou SessionWindow.

Método: deaccumulateState()

O método deaccumulateState() recalcula o estado com base no estado anterior e no estado de um salto. O Stream Analytics invoca este método quando um conjunto de eventos sai de HoppingWindow.

Método: computeResult()

O computeResult() método devolve o resultado agregado com base no estado atual. O Stream Analytics chama este método no final de uma janela temporal (TumblingWindow, HoppingWindow, SlidingWindow, ou SessionWindow).

Rever os tipos de dados de entrada e saída suportados

Agregados definidos pelo utilizador em JavaScript usam as mesmas conversões de tipos de entrada e saída que as funções definidas pelo utilizador (UDF) do JavaScript. Para o mapeamento completo entre os tipos de dados Stream Analytics e os tipos de dados JavaScript, consulte a secção Stream Analytics e conversão de tipos JavaScript do Integrar UDFs JavaScript.

Adicionar uma UDA de JavaScript no portal do Azure

Nesta secção, cria uma UDA que calcula uma média ponderada no tempo. Para criar um UDA em JavaScript num trabalho existente de Stream Analytics, siga estes passos:

  1. Inicie sessão no portal do Azure e aceda ao seu trabalho do Stream Analytics.

  2. Em Topologia da tarefa, selecione Funções.

  3. Selecione Adicionar e depois selecione JavaScript UDA.

  4. Na página Nova função, aparece no editor um modelo UDA por defeito.

  5. Introduza TWA como o alias da função e depois substitua a implementação da função pelo seguinte código:

    // Sample UDA which calculates the time-weighted average of incoming values.
function main() {
    this.init = function () {
        this.totalValue = 0.0;
        this.totalWeight = 0.0;
    }

    this.accumulate = function (value, timestamp) {
        this.totalValue += value.level * value.weight;
        this.totalWeight += value.weight;

    }

    // Uncomment the following block for an AccumulateDeaccumulate implementation.
    /*
    this.deaccumulate = function (value, timestamp) {
        this.totalValue -= value.level * value.weight;
        this.totalWeight -= value.weight;
    }

    this.deaccumulateState = function (otherState){
        this.totalValue -= otherState.totalValue;
        this.totalWeight -= otherState.totalWeight;
    }
    */

    this.computeResult = function () {
        if(this.totalValue == 0) {
            result = 0;
        }
        else {
            result = this.totalValue/this.totalWeight;
        }
        return result;
    }
}

  6. Selecione Guardar. O seu UDA aparece na lista de funções.

  7. Selecione a nova função TWA para rever a sua definição.

Invocar um UDA de JavaScript numa consulta do Stream Analytics

No portal do Azure, abra o seu trabalho e edite a consulta. Chama a TWA() função com o prefixo obrigatório uda. . Por exemplo:

WITH value AS
(
    SELECT
    NoiseLevelDB as level,
    DurationSecond as weight
FROM
    [YourInputAlias] TIMESTAMP BY EntryTime
)
SELECT
    System.Timestamp as ts,
    uda.TWA(value) as NoiseDoseTWA
FROM value
GROUP BY TumblingWindow(minute, 5)

Teste a consulta com a UDA

Crie um ficheiro JSON local com o seguinte conteúdo, carregue o ficheiro como entrada de exemplo para o seu trabalho de Stream Analytics e depois teste a consulta anterior:

[
  {"EntryTime": "2017-06-10T05:01:00-07:00", "NoiseLevelDB": 80, "DurationSecond": 22.0},
  {"EntryTime": "2017-06-10T05:02:00-07:00", "NoiseLevelDB": 81, "DurationSecond": 37.8},
  {"EntryTime": "2017-06-10T05:02:00-07:00", "NoiseLevelDB": 85, "DurationSecond": 26.3},
  {"EntryTime": "2017-06-10T05:03:00-07:00", "NoiseLevelDB": 95, "DurationSecond": 13.7},
  {"EntryTime": "2017-06-10T05:03:00-07:00", "NoiseLevelDB": 88, "DurationSecond": 10.3},
  {"EntryTime": "2017-06-10T05:05:00-07:00", "NoiseLevelDB": 103, "DurationSecond": 5.5},
  {"EntryTime": "2017-06-10T05:06:00-07:00", "NoiseLevelDB": 99, "DurationSecond": 23.0},
  {"EntryTime": "2017-06-10T05:07:00-07:00", "NoiseLevelDB": 108, "DurationSecond": 1.76},
  {"EntryTime": "2017-06-10T05:07:00-07:00", "NoiseLevelDB": 79, "DurationSecond": 17.9},
  {"EntryTime": "2017-06-10T05:08:00-07:00", "NoiseLevelDB": 83, "DurationSecond": 27.1},
  {"EntryTime": "2017-06-10T05:09:00-07:00", "NoiseLevelDB": 91, "DurationSecond": 17.1},
  {"EntryTime": "2017-06-10T05:09:00-07:00", "NoiseLevelDB": 115, "DurationSecond": 7.9},
  {"EntryTime": "2017-06-10T05:09:00-07:00", "NoiseLevelDB": 80, "DurationSecond": 28.3},
  {"EntryTime": "2017-06-10T05:10:00-07:00", "NoiseLevelDB": 55, "DurationSecond": 18.2},
  {"EntryTime": "2017-06-10T05:10:00-07:00", "NoiseLevelDB": 93, "DurationSecond": 25.8},
  {"EntryTime": "2017-06-10T05:11:00-07:00", "NoiseLevelDB": 83, "DurationSecond": 11.4},
  {"EntryTime": "2017-06-10T05:12:00-07:00", "NoiseLevelDB": 89, "DurationSecond": 7.9},
  {"EntryTime": "2017-06-10T05:15:00-07:00", "NoiseLevelDB": 112, "DurationSecond": 3.7},
  {"EntryTime": "2017-06-10T05:15:00-07:00", "NoiseLevelDB": 93, "DurationSecond": 9.7},
  {"EntryTime": "2017-06-10T05:18:00-07:00", "NoiseLevelDB": 96, "DurationSecond": 3.7},
  {"EntryTime": "2017-06-10T05:20:00-07:00", "NoiseLevelDB": 108, "DurationSecond": 0.99},
  {"EntryTime": "2017-06-10T05:20:00-07:00", "NoiseLevelDB": 113, "DurationSecond": 25.1},
  {"EntryTime": "2017-06-10T05:22:00-07:00", "NoiseLevelDB": 110, "DurationSecond": 5.3}
]

Conteúdo relacionado

  • Last updated on