WASM-modules bouwen voor gegevensstromen

Ontwikkelomgeving:

• Visual Studio Code
• (Optioneel) RedHat YAML-extensie voor VS Code
• Azure IoT-bewerkingen Gegevensstroom-extensie voor VS Code.
• CodeLLDB-extensie voor VS Code om foutopsporing van WASM-modules in te schakelen
• Docker

Docker-installatiekopieën:

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. Installeren met npx @azure-tools/dataflow-dev of wereldwijd met npm install -g @azure-tools/dataflow-dev.
• Azure CLI
• Docker

Docker-installatiekopieën:

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

Als u met Rust ontwikkelt, installeert u de Rust-toolchain en voegt u het WASI Preview 2-target toe.

# 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

Als u ontwikkelt met behulp van Python, moet u ervoor zorgen dat u Python 3.8 of hoger hebt geïnstalleerd, samen met het hulpprogramma componentize-py voor het verpakken van Python code als WASM-modules:

# 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

In dit voorbeeld wordt een voorbeeldwerkruimte gebruikt die alle benodigde resources bevat om een grafiektoepassing lokaal te bouwen en uit te voeren met behulp van de VS Code-extensie.

De voorbeeldwerkruimte openen in VS Code

Kloon de opslagplaats Explore IoT Operations als u dat nog niet hebt gedaan.

Open de map samples/wasm in Visual Studio Code door Bestand > Map Openen te selecteren en naar de map samples/wasm te navigeren.

De operators bouwen

Druk op Ctrl+Shift+P om het opdrachtenpalet te openen en te zoeken naar Azure IoT-bewerkingen: Alle operators bouwen. Selecteer release als de buildmodus.

Met deze opdracht worden alle operators in de werkruimte gebouwd en worden .wasm bestanden in de operators map gemaakt. U gebruikt de .wasm bestanden om de grafiektoepassing lokaal uit te voeren.

De grafiektoepassing lokaal uitvoeren

Als u de lokale uitvoeringsomgeving wilt starten, drukt u op Ctrl+Shift+P om het opdrachtenpalet te openen en te zoeken naar Azure IoT-bewerkingen: Ontwikkelomgeving starten. Selecteer release als de uitvoeringsmodus.

Wanneer de lokale uitvoeringsomgeving wordt uitgevoerd, drukt u op Ctrl+Shift+P om het opdrachtenpalet te openen en te zoeken naar Azure IoT-bewerkingen: Application Graph uitvoeren. Selecteer release als de uitvoeringsmodus. Met deze opdracht wordt de grafiektoepassing lokaal uitgevoerd met behulp van de lokale uitvoeringsomgeving met het graph.dataflow.yaml bestand in de werkruimte.

Het leest ook van hostapp.env.list om de omgevingsvariabele TK_CONFIGURATION_PARAMETERS in te stellen voor configuratieparameters van de gegevensstroomoperator.

Wanneer u wordt gevraagd om invoergegevens, selecteert u de data-and-images map in de werkruimte. Deze map bevat de invoergegevensbestanden voor de grafiektoepassing, inclusief temperatuur- en vochtigheidsgegevens en enkele afbeeldingen voor de momentopnamemodule.

Wacht totdat u een VS Code-melding ziet dat de logboeken gereed zijn: Log files for the run can be found at ...\wasm\data-and-images\output\logs.

De uitvoer bevindt zich in de output map onder de data-and-images map. U kunt de output map in de werkruimte openen om de uitvoerbestanden weer te geven. Het .txt bestand met de datum en tijd voor bestandsnaam bevat de verwerkte gegevens en het ziet er als volgt uit:

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

In de uitvoer ziet u dat de grafiektoepassing de invoergegevens heeft verwerkt en de uitvoer heeft gegenereerd. De uitvoer bevat temperatuur- en vochtigheidsgegevens en de objecten die in de afbeeldingen zijn gedetecteerd.

Kloon de opslagplaats Explore IoT Operations als u dat nog niet hebt gedaan:

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

De operators bouwen

Navigeer naar de map van uw grafiektoepassing en bouw alle operators:

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

Met --app de vlag geeft u de map op die het graph.dataflow.yaml bestand en de operators map bevat. De CLI bouwt alle operators in de werkruimte en maakt .wasm bestanden.

De grafiektoepassing lokaal uitvoeren

Start de lokale Docker-uitvoeringsomgeving en voer vervolgens de grafiek uit:

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

Met test de opdracht wordt de grafiek uitgevoerd die is gedefinieerd in het bestand van .test.yaml de testcase, worden de invoergegevens ingevoerd en wordt de uitvoer vergeleken met de verwachte resultaten. De voorbeeldopslagplaats bevat meerdere testscenario's, zoals:

Alle tests uitvoeren:

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

Als u de Rust- of Python-hulpprogramma's gebruikt die lokaal zijn geïnstalleerd, zijn er geen voorbeelden klaar om te worden uitgevoerd. U moet uw eigen grafiektoepassing en -operators maken en deze vervolgens lokaal bouwen en uitvoeren. Volg de instructies in de volgende sectie om een nieuwe grafiektoepassing te maken met aangepaste WASM-modules.

In plaats van een bestaande voorbeeldwerkruimte te gebruiken, maakt u een volledig nieuwe werkruimte. Met dit proces leert u hoe u een nieuwe grafiektoepassing maakt en de operators in Python en Rust programmeert.

Een nieuw grafiektoepassingsproject maken in Python

Druk op Ctrl+Shift+P om het opdrachtenpalet van VS Code te openen en te zoeken naar Azure IoT-bewerkingen: Toepassing maken:

1. Selecteer voor de map een map waarin u het project wilt maken. U kunt een nieuwe map voor dit project maken.
2. Voer de my-graph naam in.
3. Selecteer Python als taal.
4. Selecteer Kaart als het type.
5. Voer de map naam in.

U hebt nu een nieuwe VS Code-werkruimte met de basisprojectstructuur en startersbestanden. De startersbestanden bevatten het graph.dataflow.yaml bestand en de broncode van de kaartoperatorsjabloon.

Voeg Python code toe voor de module kaartoperator

Open het operators/map/map.py bestand en vervang de inhoud door de volgende code om een binnenkomende temperatuurwaarde van Fahrenheit te converteren naar 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

Zorg ervoor dat Docker wordt uitgevoerd. Druk vervolgens op Ctrl+Shift+P om het opdrachtenpalet te openen en te zoeken naar Azure IoT-bewerkingen: Alle operators bouwen. Maak een releasemodule .

Het buildproces plaatst het map.wasm bestand voor de map operator in de operators/map/bin/release map.

Rust-code toevoegen voor de filteroperatormodule

Maak een nieuwe operator door op Ctrl+Shift+P te drukken om het opdrachtenpalet te openen en te zoeken naar Azure IoT-bewerkingen: Operator maken:

1. Selecteer Rust als taal.
2. Selecteer Filter als operatortype.
3. Voer de filter naam in.

Open het operators/filter/src/lib.rs bestand en vervang de inhoud door de volgende code om waarden te filteren waarbij de temperatuur hoger is dan 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)
   }
}

Open het operators/filter/Cargo.toml bestand en voeg de volgende afhankelijkheden toe:

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

Zorg ervoor dat Docker wordt uitgevoerd. Druk vervolgens op Ctrl+Shift+P om het opdrachtenpalet te openen en te zoeken naar Azure IoT-bewerkingen: Alle operators bouwen. Maak een releasemodule .

Het buildproces plaatst het filter.wasm bestand voor de filter operator in de operators/filter/bin/release map.

Als u nieuwe operators wilt maken en bouwen met de dataflow-dev CLI, gebruikt u een van de voorbeeldprojecten als sjabloon en maakt u nieuwe operatorbronbestanden in de operators map. Gebruik vervolgens de dataflow-dev build opdracht om de operators te bouwen en de .wasm bestanden te genereren:

• Zie de voorbeelden in de explore-iot-operations/samples/wasm map voor Rust-projecten.
• Zie de voorbeelden in de map explore-iot-operations/samples/wasm-python voor Python projecten.

Maak een temperatuurconversieprogramma dat Fahrenheit transformeert naar Celsius.

Als u Rust gebruikt:

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

Maak .cargo/config.toml om het SDK-register te configureren.

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

[build]
target = "wasm32-wasip2"

Cargo.tomlbewerken:

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

Schrijven 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))
}

Voer de volgende opdracht uit om de module te bouwen:

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

U kunt een containerbuild gebruiken voor CI/CD of wanneer u geen lokale Rust-hulpprogrammaketen wilt instellen:

# 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 moet gelijk zijn aan uw crate-naam van Cargo.toml. Zie Rust Docker Builder-documenten voor meer opties.

Als u Python gebruikt:

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

Maken 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

Voer de volgende opdracht uit om de module te bouwen:

# 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

U kunt een containerbuild gebruiken voor CI/CD of wanneer u geen lokale Python hulpprogrammaketen wilt instellen:

# 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 moet overeenkomen met uw Python bestandsnaam (zonder .py). --app-type moet overeenkomen met het operatortype (map, filter, branchenzovoort). Zie Python Docker Builder-documenten voor meer opties.

Open het graph.dataflow.yaml bestand en vervang de inhoud door de volgende code:

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

Kopieer de data map met de voorbeeldgegevens uit de gekloonde opslagplaats explore-iot-operations\samples\wasm\data met voorbeelden naar de huidige werkruimte. De data map bevat drie JSON-bestanden met voorbeeldtemperatuurgegevens voor invoer.

Als u de lokale uitvoeringsomgeving eerder hebt gestopt, drukt u op Ctrl+Shift+P om het opdrachtenpalet te openen en te zoeken naar Azure IoT-bewerkingen: Ontwikkelomgeving starten. Selecteer release als de uitvoeringsmodus.

Druk op Ctrl+Shift+P om het opdrachtenpalet te openen en zoek naar Azure IoT-bewerkingen: Application Graph uitvoeren:

1. Selecteer het graph.dataflow.yaml grafiekbestand.
2. Selecteer release als de uitvoeringsmodus.
3. Selecteer de data map die u naar de werkruimte hebt gekopieerd.

De DevX-container wordt gestart om de grafiek uit te voeren. Het verwerkte resultaat wordt opgeslagen in de data/output map. Het tekstbestand in de output map bevat de verwerkte gegevens met de temperatuur die is geconverteerd naar Celsius en gefilterd op basis van de drempelwaarde.

Zie Deploy WASM-modules en gegevensstroomgrafieken voor meer informatie over het implementeren van uw aangepaste WASM-modules en -grafieken in uw Azure IoT-bewerkingen-exemplaar.

Als u uw grafiektoepassing lokaal wilt uitvoeren met de dataflow-dev CLI, gebruikt u de dataflow-dev test opdracht met een testcase die verwijst naar uw grafiekconfiguratie en invoergegevens. U kunt een nieuw YAML-testcasebestand maken in de test-runner/tests map die verwijst naar uw graph.dataflow.yaml en invoergegevensbestanden. Voer vervolgens de test uit met de volgende opdracht:

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

Bekijk de voorbeeldbestanden voor testcases in de test-runner/tests map voor naslaginformatie over het structuren van uw YAML-bestand voor testcases. U kunt zowel Python- als Rust-modules testen met het CLI-hulpprogramma door te verwijzen naar de juiste grafiekconfiguratie en invoergegevens in uw testcasebestand.

Als u standaardhulpprogramma's gebruikt die lokaal zijn geïnstalleerd, is er geen inbegrepen hulpprogramma voor het uitvoeren van de grafiektoepassing met uw aangepaste WASM-modules in uw lokale omgeving.