Implementasikan agregat yang didefinisikan pengguna JavaScript di Azure Stream Analytics

Azure Stream Analytics mendukung agregat yang didefinisikan pengguna (UDA) yang ditulis dalam JavaScript sehingga Anda dapat mengimplementasikan logika bisnis berstatus kompleks. Dengan UDA, Anda memiliki kendali penuh atas struktur data status, akumulasi status, deakumulasi status, dan perhitungan hasil agregat.

Gunakan JavaScript UDA ketika fungsi agregat bawaan tidak memenuhi kebutuhan Anda dan Anda ingin menggabungkan event berjendela dengan algoritma Anda sendiri.

Artikel ini menunjukkan cara membuat UDA dan cara memanggilnya dengan operasi berbasis jendela dalam kueri Stream Analytics.

Prasyarat

Sebelum memulai, pastikan Anda memiliki:

Pekerjaan Azure Stream Analytics yang sudah ada. Jika Anda belum memilikinya, lihat Quickstart: Membuat pekerjaan Stream Analytics menggunakan portal Azure.
Familiarisasi dengan fungsi jendela yang Anda gunakan bersama agregat.

Pilih tipe agregat yang ditentukan pengguna JavaScript

Agregat yang ditentukan pengguna berjalan di atas spesifikasi jendela waktu untuk menggabungkan peristiwa di jendela tersebut dan menghasilkan satu nilai hasil. Stream Analytics mendukung dua jenis antarmuka UDA: AccumulateOnly dan AccumulateDeaccumulate. Kedua jenis ini bekerja dengan jendela tumbling, hopping, sliding, dan session. Pilih tipe berdasarkan algoritma yang Anda gunakan.

Agregat AccumulateDeaccumulate berkinerja lebih baik daripada agregat AccumulateOnly jika Anda menggunakannya dengan jendela hopping, sliding, dan session, karena Stream Analytics dapat menghapus event dari status tersebut daripada menghitungnya ulang.

"Hanya Akumulasi mengagregasi."

Agregat AccumulateOnly hanya dapat mengumpulkan event baru ke dalam statusnya. Algoritma ini tidak memungkinkan deakumulasi nilai. Pilih tipe ini ketika Anda tidak dapat menghapus informasi suatu peristiwa dari nilai status. Kode berikut adalah template JavaScript untuk agregat 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;
    }
}

Fungsi AccumulateDeaccumulate

AkumulasiAkumulasi agregat deakumulasi nilai yang sebelumnya terkumpul dari negara bagian. Misalnya, Anda dapat menghapus pasangan kunci-nilai dari daftar nilai peristiwa atau mengurangi nilai dari agregat jumlah. Kode berikut adalah template JavaScript untuk agregat 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;
    }
}

Pahami deklarasi fungsi JavaScript

Deklarasi objek Function mendefinisikan setiap UDA JavaScript. Daftar berikut menjelaskan elemen utama dalam definisi UDA.

Alias fungsi

Alias fungsi adalah pengenal UDA. Saat Anda memanggil UDA dalam kueri Stream Analytics, selalu gunakan alias dengan prefiks uda..

Jenis Fungsi

Untuk UDA, atur tipe fungsi ke JavaScript UDA.

Jenis output

Atur tipe output ke tipe spesifik yang didukung oleh pekerjaan Stream Analytics, atau ke Any jika Anda ingin menangani tipe tersebut dalam kueri Anda.

Nama fungsi

Nama objek Function. Nama fungsi harus sesuai dengan alias UDA.

Metode: init()

Metode init() ini menginisialisasi keadaan agregat. Stream Analytics memanggil metode ini saat jendela dimulai.

Metode: akumulasi()

Metode ini accumulate() menghitung status UDA berdasarkan status sebelumnya dan nilai event saat ini. Stream Analytics memanggil metode ini ketika sebuah event memasuki jendela waktu (TumblingWindow, HoppingWindow, SlidingWindow, atau SessionWindow).

Metode: deakumulasi()

Metode ini deaccumulate() menghitung ulang status berdasarkan status sebelumnya dan nilai peristiwa saat ini. Stream Analytics memanggil metode ini ketika sebuah kejadian keluar dari SlidingWindow atau SessionWindow.

Metode: deaccumulateState()

Metode deaccumulateState() menghitung ulang status berdasarkan status sebelumnya dan status lompatan. Stream Analytics memanggil metode ini ketika sekumpulan peristiwa keluar dari HoppingWindow.

Metode: computeResult()

Metode computeResult() mengembalikan hasil agregasi berdasarkan status saat ini. Stream Analytics memanggil metode ini di akhir jendela waktu (TumblingWindow, HoppingWindow, SlidingWindow, atau SessionWindow).

Tinjau tipe data input dan output yang didukung

Agregat yang didefinisikan pengguna JavaScript menggunakan konversi tipe input dan output yang sama dengan fungsi pengguna JavaScript (UDF). Untuk pemetaan lengkap antara tipe data Stream Analytics dan tipe data JavaScript, lihat bagian konversi tipe Stream Analytics dan JavaScript di Integrasi JavaScript UDFs.

Tambahkan JavaScript UDA di portal Azure

Di bagian ini, Anda membuat UDA yang menghitung rata-rata tertimbang waktu. Untuk membuat JavaScript UDA di pekerjaan Stream Analytics yang sudah ada, ikuti langkah-langkah berikut:

Masuk ke portal Azure dan buka pekerjaan Stream Analytics Anda.
Di bawah Topologi Pekerjaan, pilih Fungsi.
Pilih Tambah, lalu pilih JavaScript UDA.
Di halaman fungsi baru , template UDA default muncul di editor.

Masukkan TWA sebagai alias fungsi, lalu ganti implementasi fungsi dengan kode berikut:

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

Pilih Simpan. UDA Anda muncul di daftar fungsi.
Pilih fungsi TWA baru untuk meninjau definisinya.

Panggil JavaScript UDA dalam kueri Stream Analytics

Di portal Azure, buka pekerjaan Anda dan edit kueri. Panggil fungsi TWA() dengan awalan uda. yang wajib. Contohnya:

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)

Uji kueri dengan UDA

Buat file JSON lokal dengan konten berikut, unggah file tersebut sebagai input contoh ke pekerjaan Stream Analytics Anda, lalu uji kueri sebelumnya:

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

Pengantar Azure Stream Analytics
Mulai menggunakan Azure Stream Analytics
Menskalakan pekerjaan Azure Stream Analytics
Referensi bahasa kueri Azure Stream Analytics
Referensi REST API manajemen Azure Stream Analytics
Halaman tanya jawab Microsoft Q&A untuk Azure Stream Analytics

Saran dan Komentar

Apakah halaman ini membantu?

Last updated on 2026-08-27