Skapa WASM-moduler för dataflöden

Utvecklingsmiljö:

• Visual Studio Code
• (Valfritt) RedHat YAML-tillägg för VS Code
• Azure IoT Operations Data Flow-tillägget för VS Code.
• CodeLLDB-tillägget för VS Code för att aktivera felsökning av WASM-moduler
• Docker

Docker-avbildningar:

docker pull mcr.microsoft.com/azureiotoperations/processor-app:1.1.5
docker tag mcr.microsoft.com/azureiotoperations/processor-app:1.1.5 host-app

docker pull mcr.microsoft.com/azureiotoperations/devx-runtime:0.1.8
docker tag mcr.microsoft.com/azureiotoperations/devx-runtime:0.1.8 devx

docker pull mcr.microsoft.com/azureiotoperations/statestore-cli:0.0.2
docker tag mcr.microsoft.com/azureiotoperations/statestore-cli:0.0.2 statestore-cli

docker pull eclipse-mosquitto

• dataflow-dev CLI (kommandoradsgränssnitt). Installera med npx @azure-tools/dataflow-dev eller globalt med npm install -g @azure-tools/dataflow-dev.
• Azure CLI
• Docker

Docker-avbildningar:

docker pull mcr.microsoft.com/azureiotoperations/processor-app:1.1.5
docker tag mcr.microsoft.com/azureiotoperations/processor-app:1.1.5 host-app

docker pull mcr.microsoft.com/azureiotoperations/devx-runtime:0.1.8
docker tag mcr.microsoft.com/azureiotoperations/devx-runtime:0.1.8 devx

docker pull mcr.microsoft.com/azureiotoperations/statestore-cli:0.0.2
docker tag mcr.microsoft.com/azureiotoperations/statestore-cli:0.0.2 statestore-cli

docker pull eclipse-mosquitto

Om du utvecklar med Rust installerar du Rust-verktygskedjan och lägger till WASI Preview 2-målet:

# Install Rust toolchain
curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y

# Add WASM target (required for Azure IoT Operations WASM components)
rustup target add wasm32-wasip2

# Install build tools for validating and packaging WASM artifacts
cargo install wasm-tools --version '=1.201.0' --locked

Om du utvecklar med hjälp av Python kontrollerar du att du har Python 3.8 eller senare installerat, tillsammans med verktyget componentize-py för att paketera Python kod som WASM-moduler:

# Install componentize-py for building Python WASM modules
pip install "componentize-py==0.14"

# Clone the samples repo (provides WIT schemas and project structure)
git clone https://github.com/Azure-Samples/explore-iot-operations.git

I det här exemplet används en exempelarbetsyta som innehåller alla nödvändiga resurser för att skapa och köra ett grafprogram lokalt med hjälp av VS Code-tillägget.

Öppna exempelarbetsytan i VS Code

Klona lagringsplatsen Explore IoT Operations om du inte redan har gjort det.

Öppna mappen samples/wasm i Visual Studio Code genom att välja File > Öppna mapp och navigera till mappen samples/wasm.

Skapa operatorerna

Tryck på Ctrl+Shift+P för att öppna kommandopaletten och sök efter Azure IoT Operations: Skapa alla operatorer. Välj version som byggläge.

Det här kommandot skapar alla operatorer på arbetsytan och skapar .wasm filer i operators mappen. Du använder .wasm filerna för att köra grafprogrammet lokalt.

Köra grafprogrammet lokalt

Starta den lokala körningsmiljön genom att trycka på Ctrl+Shift+P för att öppna kommandopaletten och söka efter Azure IoT Operations: Starta utvecklingsmiljön. Välj släpp som körningsläge.

När den lokala körningsmiljön körs trycker du på Ctrl+Shift+P för att öppna kommandopaletten och söka efter Azure IoT Operations: Kör Application Graph. Välj släpp som körningsläge. Det här kommandot kör grafprogrammet lokalt med hjälp av den lokala körningsmiljön med graph.dataflow.yaml filen på arbetsytan.

Den läser också från hostapp.env.list för att ange miljövariabeln TK_CONFIGURATION_PARAMETERS för konfigurationsparametrar för dataflödesoperatorer.

När du uppmanas att ange indata väljer du data-and-images mappen på arbetsytan. Den här mappen innehåller indatafilerna för grafprogrammet, inklusive temperatur- och luftfuktighetsdata och några bilder för modulen ögonblicksbild.

Vänta tills du ser ett VS Code-meddelande om att loggarna är klara: Log files for the run can be found at ...\wasm\data-and-images\output\logs.

Utdata finns i output mappen under data-and-images mappen . Du kan öppna output mappen på arbetsytan för att se utdatafilerna. Filen .txt med datum och tid för filnamn innehåller bearbetade data och ser ut som i följande exempel:

{"tst":"2025-09-19T04:19:13.530381+0000","topic":"sensors","qos":0,"retain":0,"payloadlen":312,"properties":{"payload-format-indicator":1,"message-expiry-interval":10,"correlation-data":"...","user-properties":{"__ts":"001758255553528:00000:...","__protVer":"1.0","__srcId":"mqtt-source"},"content-type":"application/json"},"payload":{"temperature":[{"count":2,"max":653.888888888889,"min":204.44444444444449,"average":429.16666666666669,"last":204.44444444444449,"unit":"C","overtemp":true}],"humidity":[{"count":3,"max":85,"min":45,"average":69.666666666666671,"last":79}],"object":[{"result":"notebook, notebook computer; sliding door"}]}}

Utdata visar att grafprogrammet bearbetade indata och genererade utdata. Utdata innehåller temperatur- och luftfuktighetsdata och de objekt som identifieras i bilderna.

Klona lagringsplatsen Explore IoT Operations om du inte redan har gjort det:

git clone https://github.com/Azure-Samples/explore-iot-operations.git

Skapa operatorerna

Gå till katalogen för ditt diagramprogram och skapa alla operatorer:

cd explore-iot-operations/samples/wasm
dataflow-dev build --app .

Flaggan --app anger den katalog som innehåller graph.dataflow.yaml filen och operators mappen. CLI bygger alla operatorer på arbetsytan och genererar .wasm-filer.

Köra grafprogrammet lokalt

Starta den lokala Docker-körningsmiljön och kör sedan diagrammet:

dataflow-dev run start
dataflow-dev test --app . test-runner/tests/t03-complex-full-pipeline
dataflow-dev run stop

Kommandot test kör diagrammet som definierats i testfallets .test.yaml fil, matar indata och jämför utdata med förväntade resultat. Exempellagringsplatsen innehåller flera testscenarier, till exempel:

Så här kör du alla tester:

dataflow-dev run start
dataflow-dev test --app . test-runner/tests
dataflow-dev run stop

Om du använder Rust- eller Python-verktygen som är installerade lokalt finns det inga exempel som är redo att köras. Du måste skapa ett eget diagramprogram och operatorer och sedan skapa och köra dem lokalt. Följ anvisningarna i nästa avsnitt för att skapa ett nytt diagramprogram med anpassade WASM-moduler.

I stället för att använda en befintlig exempelarbetsyta skapar du en ny arbetsyta från grunden. Med den här processen kan du lära dig hur du skapar ett nytt grafprogram och programmerar operatorerna i Python och Rust.

Skapa ett nytt diagramprogramprojekt i Python

Tryck på Ctrl+Shift+P för att öppna kommandopaletten för VS Code och sök efter Azure IoT Operations: Skapa program:

1. För mappen väljer du en mapp där du vill skapa projektet. Du kan skapa en ny mapp för det här projektet.
2. Ange my-graph som namn.
3. Välj Python som språk.
4. Välj Mappa som typ.
5. Ange map som namn.

Nu har du en ny VS Code-arbetsyta med den grundläggande projektstrukturen och startfilerna. Startfilerna graph.dataflow.yaml innehåller filen och källkoden för kartoperatorns mall.

Lägg till Python kod för kartoperatormodulen

operators/map/map.py Öppna filen och ersätt innehållet med följande kod för att konvertera ett inkommande temperaturvärde från Fahrenheit till Celsius:

import json
from map_impl import exports
from map_impl import imports
from map_impl.imports import types

class Map(exports.Map):
    def init(self, configuration) -> bool:
        imports.logger.log(imports.logger.Level.INFO, "module4/map", "Init invoked")
        return True

    def process(self, message: types.DataModel) -> types.DataModel:
        # TODO: implement custom logic for map operator
        imports.logger.log(imports.logger.Level.INFO, "module4/map", "processing from python")

        # Ensure the input is of the expected type
        if not isinstance(message, types.DataModel_Message):
            raise ValueError("Unexpected input type: Expected DataModel_Message")

        # Extract and decode the payload
        payload_variant = message.value.payload
        if isinstance(payload_variant, types.BufferOrBytes_Buffer):
            # It's a Buffer handle - read from host
            imports.logger.log(imports.logger.Level.INFO, "module4/map", "Reading payload from Buffer")
            payload = payload_variant.value.read()
        elif isinstance(payload_variant, types.BufferOrBytes_Bytes):
            # It's already bytes
            imports.logger.log(imports.logger.Level.INFO, "module4/map", "Reading payload from Bytes")
            payload = payload_variant.value
        else:
            raise ValueError("Unexpected payload type")

        decoded = payload.decode("utf-8")

        # Parse the JSON data
        json_data = json.loads(decoded)

        # Check and update the temperature value
        if "temperature" in json_data and "value" in json_data["temperature"]:
            temp_f = json_data["temperature"]["value"]
            if isinstance(temp_f, int):
                # Convert Fahrenheit to Celsius
                temp_c = round((temp_f - 32) * 5.0 / 9.0)

                # Update the JSON data
                json_data["temperature"]["value"] = temp_c
                json_data["temperature"]["unit"] = "C"

                # Serialize the updated JSON back to bytes
                updated_payload = json.dumps(json_data).encode("utf-8")

                # Update the message payload
                message.value.payload = types.BufferOrBytes_Bytes(value=updated_payload)

        return message

Kontrollera att Docker körs. Tryck sedan på Ctrl+Shift+P för att öppna kommandopaletten och söka efter Azure IoT Operations: Skapa alla operatorer. Skapa en release-modul.

Byggprocessen placerar map.wasm filen för operatorn map i operators/map/bin/release mappen .

Lägg till Rust-kod för filteroperatormodulen

Skapa en ny operator genom att trycka på Ctrl+Shift+P för att öppna kommandopaletten och söka efter Azure IoT Operations: Skapa operator:

1. Välj Rust som språk.
2. Välj Filtrera som operatortyp.
3. Ange filter som namn.

operators/filter/src/lib.rs Öppna filen och ersätt innehållet med följande kod för att filtrera bort värden där temperaturen är över 500°C:

mod filter_operator {
    use wasm_graph_sdk::macros::filter_operator;
    use serde_json::Value;

    fn filter_init(_configuration: ModuleConfiguration) -> bool {
        // Add code here to process the module init properties and module schemas from the configuration

        true
    }

    #[filter_operator(init = "filter_init")]
    fn filter(input: DataModel) -> Result<bool, Error> {

        // Extract payload from input to process
        let payload = match input {
            DataModel::Message(Message {
                payload: BufferOrBytes::Buffer(buffer),
                ..
            }) => buffer.read(),
            DataModel::Message(Message {
                payload: BufferOrBytes::Bytes(bytes),
                ..
            }) => bytes,
            _ => return Err(Error { message: "Unexpected input type".to_string() }),
        };

        // ... perform filtering logic here and return boolean
        if let Ok(payload_str) = std::str::from_utf8(&payload) {
            if let Ok(json) = serde_json::from_str::<Value>(payload_str) {
                if let Some(temp_c) = json["temperature"]["value"].as_i64() {
                    // Return true if temperature is above 500°C
                    return Ok(temp_c > 500);
                }
            }
        }

        Ok(false)
   }
}

operators/filter/Cargo.toml Öppna filen och lägg till följande beroenden:

[dependencies]
# ...
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"

Kontrollera att Docker körs. Tryck sedan på Ctrl+Shift+P för att öppna kommandopaletten och söka efter Azure IoT Operations: Skapa alla operatorer. Skapa en release-modul.

Byggprocessen placerar filter.wasm filen för operatorn filter i operators/filter/bin/release mappen .

Om du vill skapa nya operatorer med dataflow-dev CLI, använder du ett av exempelprojekten som en mall och skapar nya operatörskällfiler i operators mapp. dataflow-dev build Använd sedan kommandot för att skapa operatorerna och generera .wasm filerna:

• Information om Rust-projekt finns i exemplen explore-iot-operations/samples/wasm i mappen .
• För Python projekt, se exemplen i mappen explore-iot-operations/samples/wasm-python.

Skapa en temperaturkonverterare som omvandlar Fahrenheit till Celsius.

Om du använder Rust:

cargo new --lib temperature-converter
cd temperature-converter

Skapa .cargo/config.toml för att konfigurera SDK-registret:

[registries]
aio-wg = { index = "sparse+https://pkgs.dev.azure.com/azure-iot-sdks/iot-operations/_packaging/preview/Cargo/index/" }

[build]
target = "wasm32-wasip2"

Redigera Cargo.toml:

[package]
name = "temperature-converter"
version = "0.1.0"
edition = "2021"

[dependencies]
wit-bindgen = "0.22"
wasm_graph_sdk = { version = "=1.1.3", registry = "aio-wg" }
serde = { version = "1", default-features = false, features = ["derive"] }
serde_json = { version = "1", default-features = false, features = ["alloc"] }

[lib]
crate-type = ["cdylib"]

Skriv src/lib.rs:

use serde_json::{json, Value};

use wasm_graph_sdk::logger::{self, Level};
use wasm_graph_sdk::macros::map_operator;

fn fahrenheit_to_celsius_init(_configuration: ModuleConfiguration) -> bool {
    logger::log(Level::Info, "temperature-converter", "Init invoked");
    true
}

#[map_operator(init = "fahrenheit_to_celsius_init")]
fn fahrenheit_to_celsius(input: DataModel) -> Result<DataModel, Error> {
    let DataModel::Message(mut result) = input else {
        return Err(Error {
            message: "Unexpected input type".to_string(),
        });
    };

    let payload = &result.payload.read();
    if let Ok(data_str) = std::str::from_utf8(payload) {
        if let Ok(mut data) = serde_json::from_str::<Value>(data_str) {
            if let Some(temp) = data["temperature"]["value"].as_f64() {
                let celsius = (temp - 32.0) * 5.0 / 9.0;
                data["temperature"] = json!({
                    "value_celsius": celsius,
                    "original_fahrenheit": temp
                });

                if let Ok(output_str) = serde_json::to_string(&data) {
                    result.payload = BufferOrBytes::Bytes(output_str.into_bytes());
                }
            }
        }
    }

    Ok(DataModel::Message(result))
}

Skapa modulen genom att köra:

cargo build --release --target wasm32-wasip2
cp target/wasm32-wasip2/release/temperature_converter.wasm .

Du kan använda en containerbaserad version för CI/CD eller när du inte vill konfigurera en lokal Rust-verktygskedja:

# Release build
docker run --rm -v "$(pwd):/workspace" \
  ghcr.io/azure-samples/explore-iot-operations/rust-wasm-builder \
  --app-name temperature-converter

# Debug build (includes symbols)
docker run --rm -v "$(pwd):/workspace" \
  ghcr.io/azure-samples/explore-iot-operations/rust-wasm-builder \
  --app-name temperature-converter --build-mode debug

--app-name måste matcha ditt crate-namn från Cargo.toml. Mer information finns i Rust Docker builder docs.

Om du använder Python:

cd explore-iot-operations/samples/wasm-python/operators/map

Skapa temperature_converter.py:

import json
from map_impl import exports
from map_impl import imports
from map_impl.imports import types

class Map(exports.Map):
    def init(self, configuration) -> bool:
        imports.logger.log(imports.logger.Level.INFO, "temperature-converter", "Init invoked")
        return True

    def process(self, message: types.DataModel) -> types.DataModel:
        if not isinstance(message, types.DataModel_Message):
            raise ValueError("Unexpected input type: Expected DataModel_Message")

        payload_variant = message.value.payload
        if isinstance(payload_variant, types.BufferOrBytes_Buffer):
            payload = payload_variant.value.read()
        elif isinstance(payload_variant, types.BufferOrBytes_Bytes):
            payload = payload_variant.value
        else:
            raise ValueError("Unexpected payload type")

        decoded = payload.decode("utf-8")
        data = json.loads(decoded)

        if "temperature" in data and "value" in data["temperature"]:
            temp_f = data["temperature"]["value"]
            if isinstance(temp_f, (int, float)):
                temp_c = (temp_f - 32) * 5.0 / 9.0
                data["temperature"]["value"] = temp_c
                data["temperature"]["unit"] = "C"
                updated_payload = json.dumps(data).encode("utf-8")
                message.value.payload = types.BufferOrBytes_Bytes(value=updated_payload)

        return message

Skapa modulen genom att köra:

# Generate Python bindings from WIT schemas
componentize-py -d ../../schema -w map-impl bindings ./

# Build the WASM module
componentize-py -d ../../schema -w map-impl componentize temperature_converter -o temperature_converter.wasm

# Verify build
file temperature_converter.wasm  # Should show: WebAssembly (wasm) binary module

Du kan använda en containerbaserad version för CI/CD eller när du inte vill konfigurera en lokal Python verktygskedja:

# Release build
docker run --rm -v "$(pwd):/workspace" \
  ghcr.io/azure-samples/explore-iot-operations/python-wasm-builder \
  --app-name temperature_converter --app-type map

# Debug build (includes symbols)
docker run --rm -v "$(pwd):/workspace" \
  ghcr.io/azure-samples/explore-iot-operations/python-wasm-builder \
  --app-name temperature_converter --app-type map --build-mode debug

--app-name måste matcha ditt Python filnamn (utan .py). --app-type måste matcha operatortypen (map, filter, branchosv.). Mer information finns i Python Docker builder docs.

graph.dataflow.yaml Öppna filen och ersätt innehållet med följande kod:

metadata:
    $schema: "https://www.schemastore.org/aio-wasm-graph-config-1.0.0.json"
    name: "Temperature Monitoring"
    description: "A graph that converts temperature from Fahrenheit to Celsius, if temperature is above 500°C, then sends the processed data to the sink."
    version: "1.0.0"
    vendor: "Microsoft"
moduleRequirements:
    apiVersion: "1.1.0"
    runtimeVersion: "1.1.0"
operations:
  - operationType: source
    name: source
  - operationType: map
    name: map
    module: map
  - operationType: filter
    name: filter
    module: filter
  - operationType: sink
    name: sink
connections:
  - from:
      name: source
    to:
      name: map
  - from:
      name: map
    to:
      name: filter
  - from:
      name: filter
    to:
      name: sink

Kopiera mappen data som innehåller exempeldata från den klonade exempellagringsplatsen explore-iot-operations\samples\wasm\data till den aktuella arbetsytan. Mappen data innehåller tre JSON-filer med exempeldata för indatatemperatur.

Om du tidigare stoppade den lokala körningsmiljön trycker du på Ctrl+Shift+P för att öppna kommandopaletten och söka efter Azure IoT Operations: Starta utvecklingsmiljön. Välj släpp som körningsläge.

Tryck på Ctrl+Shift+P för att öppna kommandopaletten och sök efter Azure IoT Operations: Kör Application Graph:

1. Välj graph.dataflow.yaml graffilen.
2. Välj släpp som körningsläge.
3. Välj mappen som data du kopierade till arbetsytan.

DevX-containern startas för att köra diagrammet. Det bearbetade resultatet sparas i data/output mappen. Textfilen i output mappen innehåller bearbetade data med temperaturen konverterad till Celsius och filtrerad baserat på tröskelvärdet.

Information om hur du distribuerar dina anpassade WASM-moduler och diagram till din Azure IoT Operations-instans finns i Distribuera WASM-moduler och dataflödesdiagram.

Om du vill köra grafprogrammet lokalt med dataflödes-dev CLI använder du dataflow-dev test kommandot med ett testfall som pekar på grafkonfigurationen och indata. Du kan skapa en ny YAML-fil för testfall i test-runner/tests mappen som refererar till dina graph.dataflow.yaml och indatafiler. Kör sedan testet med följande kommando:

dataflow-dev run start
dataflow-dev test --app . test-runner/tests/my-test-case
dataflow-dev run stop

Titta på exempelfilerna för testfall i test-runner/tests mappen för att se hur du strukturerar YAML-filen för testfall. Du kan testa både Python- och Rust-moduler med CLI-verktyget genom att peka på lämplig grafkonfiguration och mata in data i testfallsfilen.

Om du använder standardverktyg som installerats lokalt finns det inget inkluderat verktyg för att köra grafprogrammet med dina anpassade WASM-moduler i din lokala miljö.