Agregacje języka JavaScript zdefiniowane przez użytkownika w usłudze Azure Stream Analytics

Usługa Azure Stream Analytics obsługuje agregacje zdefiniowane przez użytkownika (UDA) napisane w języku JavaScript, dzięki czemu można zaimplementować złożoną logikę biznesową stanową. W obrębie UDA masz pełną kontrolę nad strukturą danych stanu, akumulacją stanu, dekomulacji stanu i agregacją obliczeń wyników. W tym artykule przedstawiono dwa różne interfejsy UDA języka JavaScript, kroki tworzenia uda oraz sposób używania rozwiązania UDA z operacjami opartymi na oknach w zapytaniu usługi Stream Analytics.

Agregacje zdefiniowane przez użytkownika w języku JavaScript

Agregacja zdefiniowana przez użytkownika jest używana na podstawie specyfikacji przedziału czasu w celu agregowania zdarzeń w tym oknie i generowania pojedynczej wartości wyniku. Istnieją dwa typy interfejsów UDA, które usługa Stream Analytics obsługuje obecnie: AccumulateOnly i AccumulateDeaccumulate. Oba typy UDA mogą być używane przez wirowanie, przeskoki, przesuwanie i okno sesji. Funkcja AccumulateDeaccumulate UDA działa lepiej niż akumulacja UDA, gdy jest używana razem z przeskokami, przesuwaniem i oknem sesji. Wybierasz jeden z dwóch typów na podstawie używanego algorytmu.

Agregacje skumulowane

Agregacje skumulowane mogą gromadzić tylko nowe zdarzenia w swoim stanie. Algorytm nie zezwala na deakkumulację wartości. Wybierz ten typ agregacji, gdy zdekumulowanie informacji o zdarzeniu z wartości stanu jest niemożliwe do zaimplementowania. Poniżej znajduje się szablon języka JavaScript dla agregacji akumulowanych:

// 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;
    }
}

Akumulacja agregacji skumulowanych

AggregateDeaccumulate umożliwiają deakkumulację poprzedniej skumulowanej wartości ze stanu, na przykład usunięcie pary klucz-wartość z listy wartości zdarzeń lub odejmowanie wartości z stanu agregacji sumy. Poniżej znajduje się szablon języka JavaScript dla agregacji 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;
    }
}

UDA — deklaracja funkcji JavaScript

Każdy identyfikator UDA języka JavaScript jest definiowany przez deklarację obiektu funkcji. Poniżej przedstawiono główne elementy w definicji UDA.

Alias funkcji

Alias funkcji jest identyfikatorem UDA. Podczas wywoływania w zapytaniu usługi Stream Analytics zawsze używaj aliasu UDA razem z prefiksem "uda".

Typ funkcji

W przypadku interfejsu UDA typ funkcji powinien mieć wartość JavaScript UDA.

Typ danych wyjściowych

Określony typ obsługiwany przez zadanie usługi Stream Analytics lub "Dowolny", jeśli chcesz obsłużyć typ zapytania.

Nazwa funkcji

Nazwa tego obiektu funkcji. Nazwa funkcji powinna być zgodna z aliasem UDA.

Metoda — init()

Metoda init() inicjuje stan agregacji. Ta metoda jest wywoływana po uruchomieniu okna.

Method — accumulate()

Metoda accumulate() oblicza stan UDA na podstawie poprzedniego stanu i bieżących wartości zdarzeń. Ta metoda jest wywoływana, gdy zdarzenie wchodzi w przedział czasu (TUMBLINGWINDOW, HOPPINGWINDOW, SLIDINGWINDOW lub SESSIONWINDOW).

Method — deaccumulate()

Metoda deaccumulate() ponownie oblicza stan na podstawie poprzedniego stanu i bieżących wartości zdarzeń. Ta metoda jest wywoływana, gdy zdarzenie opuszcza SLIDINGWINDOW lub SESSIONWINDOW.

Method — deaccumulateState()

Metoda deaccumulateState() ponownie oblicza stan na podstawie poprzedniego stanu i stanu przeskoku. Ta metoda jest wywoływana, gdy zestaw zdarzeń opuszcza HOPPINGWINDOW.

Method — computeResult()

Metoda computeResult() zwraca zagregowany wynik na podstawie bieżącego stanu. Ta metoda jest wywoływana na końcu okna czasowego (TUMBLINGWINDOW, HOPPINGWINDOW, SLIDINGWINDOW lub SESSIONWINDOW).

Obsługiwane typy danych wejściowych i wyjściowych w języku JavaScript UDA

W przypadku typów danych UDA języka JavaScript zapoznaj się z sekcją Stream Analytics i konwersją typu JavaScript integrowania funkcji UDF języka JavaScript.

Dodawanie interfejsu UDA języka JavaScript z witryny Azure Portal

Poniżej przedstawiono proces tworzenia uda z poziomu portalu. W tym przykładzie użyto tutaj średnich ważonych czasowo.

Teraz utwórzmy uda języka JavaScript w ramach istniejącego zadania usługi ASA, wykonując następujące kroki.

1. Zaloguj się do witryny Azure Portal i znajdź istniejące zadanie usługi Stream Analytics.

2. Następnie wybierz link funkcje w obszarze TOPOLOGIA ZADANIA.

3. Wybierz pozycję Dodaj , aby dodać nową funkcję.

4. W widoku Nowa funkcja wybierz pozycję JavaScript UDA jako typ funkcji, a następnie w edytorze zostanie wyświetlony domyślny szablon UDA.

5. Wypełnij ciąg "TWA" jako alias UDA i zmień implementację funkcji w następujący sposób:

   // Sample UDA which calculate 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 below for AccumulateDeaccumulate implementation
       /*
       this.deaccumulate = function (value, timestamp) {
           this.totalValue -= value.level * value.weight;
           this.totalWeight -= value.weight;
       }
   
       this.deaccumulateState = function (otherState){
           this.state -= otherState.state;
           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. Po wybraniu przycisku "Zapisz" uda zostanie wyświetlona na liście funkcji.

7. Wybierz nową funkcję "TWA", możesz sprawdzić definicję funkcji.

Wywoływanie interfejsu UDA języka JavaScript w zapytaniu ASA

W witrynie Azure Portal i otwórz zadanie, edytuj zapytanie i wywołaj funkcję TWA() przy użyciu prefiksu mandatu "uda.". Na przykład:

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

Testowanie zapytania za pomocą uda

Utwórz lokalny plik JSON z poniższą zawartością, przekaż plik do zadania usługi Stream Analytics i przetestuj powyższe zapytanie.

[
  {"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}
]

Uzyskaj pomoc

Aby uzyskać dodatkową pomoc, wypróbuj stronę pytań i odpowiedzi firmy Microsoft dotyczącą usługi Azure Stream Analytics.

Następne kroki

• Wprowadzenie do usługi Azure Stream Analytics
• Get started using Azure Stream Analytics (Rozpoczynanie pracy z usługą Azure Stream Analytics)
• Scale Azure Stream Analytics jobs (Skalowanie zadań usługi Azure Stream Analytics)
• Dokumentacja języka zapytań usługi Azure Stream Analytics
• Dokumentacja interfejsu API REST zarządzania usługą Azure Stream Analytics