Entwickeln von WebAssembly-Modulen (WASM) und Diagrammdefinitionen für Datenflussdiagramme

• Rust-Toolkette bietet cargo, rustcund die Standardbibliothek, die zum Kompilieren von Operatoren benötigt wird. Installieren mit:

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

• WASM wasm32-wasip2 Ziel, das zum Erstellen von Azure IoT Operations WASM-Komponenten erforderlich ist. Hinzufügen durch:

  rustup target add wasm32-wasip2
  

• Build-Tools stellen Dienstprogramme bereit, die von den Buildern und der KI verwendet werden, um WASM-Artefakte zu validieren und zu packen. Installieren mit:

  cargo install wasm-tools --version '=1.201.0' --locked
  

• Python 3.8 oder höher

• componentize-py generiert Bindungen und erzeugt WASM-Komponenten aus Python-Quellen, die mit den Azure IoT Operations WIT-Schemas übereinstimmen. Installieren mit:

  pip install "componentize-py==0.14"
  

Sie können über eine benutzerdefinierte Microsoft Azure DevOps-Registrierung auf das WASM Rust SDK zugreifen. Statt Umgebungsvariablen zu verwenden, konfigurieren Sie den Zugriff mit einer Arbeitsbereichskonfigurationsdatei:

# .cargo/config.toml (at your workspace root)
[registries]
aio-wg = { index = "sparse+https://pkgs.dev.azure.com/azure-iot-sdks/iot-operations/_packaging/preview/Cargo/index/" }

[net]
git-fetch-with-cli = true

Dieses Setup spiegelt das Beispiellayout bei samples/wasm/.cargo/config.toml und behält Registrierungseinstellungen in der Versionssteuerung bei.

Die Python-Entwicklung verwendet „componentize-py“ mit WIT (WebAssembly Interface Types, WebAssembly-Schnittstellentypen) für die Codegenerierung. Die WIT-Schemas definieren die Schnittstellen zwischen Ihrem Python-Code und der WASM-Laufzeit.

Rufen Sie die WIT-Schemas ab: Sie finden die erforderlichen Schemas im Azure IoT Operations-Beispiel-Repository. Klonen sie, oder laden Sie diese Schemas in Ihre Entwicklungsumgebung herunter:

# Clone the repository to access WIT schemas
git clone https://github.com/Azure-Samples/explore-iot-operations.git

Die Schemas befinden sich unter explore-iot-operations/samples/wasm-python/schema/ und enthalten Schnittstellendefinitionen für alle unterstützten Operatortypen, wie Abbildung, Filter und Verzweigung.

Sie benötigen keine andere Umgebungskonfiguration, die über die Installation der erforderlichen Komponenten und das Abrufen der WIT-Schemas hinausgeht.

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

Konfigurieren von „Cargo.toml“

Bearbeiten Sie die Cargo.toml Datei, um Abhängigkeiten für das WASM SDK und andere Bibliotheken einzuschließen:

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

[dependencies]
# WebAssembly Interface Types (WIT) code generation
wit-bindgen = "0.22"

# Azure IoT Operations WASM SDK - provides operator macros and host APIs
wasm_graph_sdk = { version = "=1.1.3", registry = "aio-wg" }

# JSON serialization/deserialization for data processing
serde = { version = "1", default-features = false, features = ["derive"] }
serde_json = { version = "1", default-features = false, features = ["alloc"] }

[lib]
# Required for WASM module compilation
crate-type = ["cdylib"]

Hinweise zu Versionen und Registrierung:

• Die SDK-Version (=1.1.3) stimmt mit den aktuellen Beispielen überein; wenn Sie sie angeheftet lassen, vermeiden Sie fehlerhafte Änderungen.
• registry = "aio-wg" entspricht dem Registriereintrag, der in .cargo/config.toml definiert ist.

Die wichtigsten Abhängigkeiten wurden erläutert:

• wit-bindgen: Generiert Rust-Bindungen aus WebAssembly Interface Types (WIT)-Definitionen, damit Ihr Code mit der WASM-Laufzeit kommunizieren kann.
• wasm_graph_sdk: Das Azure IoT Operations SDK stellt Operatormakros bereit, wie z.B. #[map_operator], und #[filter_operator] Host-APIs für Protokollierung, Metriken und Zustandsverwaltung.
• serde + serde_json: JSON-Verarbeitungsbibliotheken zum Analysieren und Generieren von Datennutzlasten. default-features = false optimiert für WASM-Größenbeschränkungen.
• crate-type = ["cdylib"]: Kompiliert die Rust-Bibliothek als C-kompatible dynamische Bibliothek, die für die WASM-Modulgenerierung erforderlich ist.

Erstellen Sie eine Python-Datei für Ihren Operator. Der Dateiname sollte ihrem Modulnamen entsprechen:

# Create your operator file
touch temperature_converter.py

Python-WASM-Module erfordern keine anderen Projektkonfigurationsdateien. Die Python-Klasse, die die Operatorschnittstelle implementiert, definiert die Modulstruktur.

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"); // one-time module init
    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(); // payload bytes from inbound message
    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 fahrenheit = (temp * 9.0 / 5.0) + 32.0; // Celsius -> Fahrenheit
                data["temperature"] = json!({
                    "value_fahrenheit": fahrenheit,
                    "original_celsius": temp
                });

                if let Ok(output_str) = serde_json::to_string(&data) {
                    // Replace payload with owned bytes so the host receives the updated JSON
                    result.payload = BufferOrBytes::Bytes(output_str.into_bytes());
                }
            }
        }
    }

    Ok(DataModel::Message(result))
}

# 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:
        # Expect a typed message
        if not isinstance(message, types.DataModel_Message):
            raise ValueError("Unexpected input type: Expected DataModel_Message")

        # Extract payload (Buffer from host or Bytes inline)
        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)

        # Convert Fahrenheit to Celsius if present
        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

# Build WASM module
cargo build --release --target wasm32-wasip2  # target required for Azure IoT Operations WASM components

# Find your module  
ls target/wasm32-wasip2/release/*.wasm

# Navigate to the map operator directory (schema lives in ../../schema)
cd explore-iot-operations/samples/wasm-python/operators/map

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

# Build 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

Das Repository für Azure IoT Operations-Beispiele verwaltet den Rust Docker Builder und enthält alle erforderlichen Abhängigkeiten. Ausführliche Dokumentation finden Sie in der Verwendung des Rust Docker Builder.

# Build release version (optimized for production)
docker run --rm -v "$(pwd):/workspace" ghcr.io/azure-samples/explore-iot-operations/rust-wasm-builder --app-name temperature-converter

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

Docker-Buildoptionen:

• --app-name: Muss Ihrem Rust-Cratenamen aus Cargo.toml entsprechen.
• --build-mode: Wählen Sie release (Standard) für optimierte Builds oder debug für Entwicklungsbuilds mit Symbolen aus.

Das Repository für Azure IoT Operations-Beispiele verwaltet den Python Docker Builder und enthält alle erforderlichen Abhängigkeiten und Schemas. Ausführliche Dokumentation finden Sie unter Python Docker Builder-Verwendung.

# Build release version (optimized for production)
docker run --rm -v "$(pwd):/workspace" ghcr.io/azure-samples/explore-iot-operations/python-wasm-builder --app-name temperature_converter --app-type map

# Build debug version (includes debugging symbols and less optimization)
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

Docker-Buildoptionen:

• --app-name: Muss ihrem Python-Dateinamen ohne die .py Erweiterung entsprechen
• --app-type: Geben Sie den Operatortyp an (map, filter, branch, usw.)
• --build-mode: Wählen Sie release (Standard) für optimierte Builds oder debug für Entwicklungsbuilds mit Symbolen aus.

Umfassende Beispiele finden Sie in den Rust-Beispielen im Beispiel-Repository. Zu den vollständigen Implementierungen gehören folgende:

• map-Operatoren: Datentransformation und Konvertierungslogik
• filter-Operatoren: Bedingte Datenverarbeitung und Validierung
• brach-Operatoren: Routing mit mehreren Pfaden basierend auf Dateninhalten
• accumulate-Operatoren: Zeitfensteraggregation und statistische Verarbeitung
• delay-Operatoren: Zeitbasierte Verarbeitungssteuerung

Die Beispiele veranschaulichen Arbeitsimplementierungen, die die vollständige Struktur für jeden Operatortyp darstellen, einschließlich der richtigen Fehlerbehandlung und Protokollierungsmuster.

Umfassende Beispiele finden Sie in den Python-Beispielen im Beispiel-Repository. Zu den vollständigen Implementierungen gehören folgende:

• map-Operatoren: Datentransformation und Konvertierungslogik
• filter-Operatoren: Bedingte Datenverarbeitung und Validierung
• brach-Operatoren: Routing mit mehreren Pfaden basierend auf Dateninhalten

Das WASM Rust SDK bietet umfassende Entwicklungstools:

Operatormakros

use wasm_graph_sdk::macros::{map_operator, filter_operator, branch_operator};
use wasm_graph_sdk::{DataModel, HybridLogicalClock};

// Map operator - transforms each data item
#[map_operator(init = "my_init_function")]
fn my_map(input: DataModel) -> Result<DataModel, Error> {
    // Transform logic here
}

// Filter operator - allows/rejects data based on predicate  
#[filter_operator(init = "my_init_function")]
fn my_filter(input: DataModel) -> Result<bool, Error> {
    // Return true to pass data through, false to filter out
}

// Branch operator - routes data to different arms
#[branch_operator(init = "my_init_function")]
fn my_branch(input: DataModel, timestamp: HybridLogicalClock) -> Result<bool, Error> {
    // Return true for "True" arm, false for "False" arm
}

Modulkonfigurationsparameter

Ihre WASM-Operatoren können Laufzeitkonfigurationsparameter über die ModuleConfiguration an die init Funktion übergebene Struktur empfangen. Sie definieren diese Parameter in der Diagrammdefinition, mit der Sie die Laufzeit anpassen können, ohne Module neu zu erstellen.

use wasm_graph_sdk::logger::{self, Level};
use wasm_graph_sdk::ModuleConfiguration;

fn my_operator_init(configuration: ModuleConfiguration) -> bool {
    // Access required parameters
    if let Some(threshold_param) = configuration.parameters.get("temperature_threshold") {
        let threshold: f64 = threshold_param.parse().unwrap_or(25.0);
        logger::log(Level::Info, "my-operator", &format!("Using threshold: {}", threshold));
    }
    
    // Access optional parameters with defaults
    let unit = configuration.parameters
        .get("output_unit")
        .map(|s| s.as_str())
        .unwrap_or("celsius");
    
    logger::log(Level::Info, "my-operator", &format!("Output unit: {}", unit));
    true
}

Ausführliche Informationen zum Definieren von Konfigurationsparametern in Diagrammdefinitionen finden Sie unter Modulkonfigurationsparameter.

Hosten von APIs

Verwenden Sie das SDK, um mit verteilten Diensten zu arbeiten:

Zustandsspeicher für persistente Daten:

use wasm_graph_sdk::state_store;

// Set value
state_store::set(key.as_bytes(), value.as_bytes(), None, None, options)?;

// Get value  
let response = state_store::get(key.as_bytes(), None)?;

// Delete key
state_store::del(key.as_bytes(), None, None)?;

Strukturierte Protokollierung:

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

logger::log(Level::Info, "my-operator", "Processing started");
logger::log(Level::Error, "my-operator", &format!("Error: {}", error));

Mit OpenTelemetry kompatible Metriken:

use wasm_graph_sdk::metrics;

// Increment counter
metrics::add_to_counter("requests_total", 1.0, Some(labels))?;

// Record histogram value
metrics::record_to_histogram("processing_duration", duration_ms, Some(labels))?;

Die Python-WASM-Entwicklung verwendet kein herkömmliches SDK. Stattdessen verwenden Sie generierte Bindungen aus WIT (WebAssembly Interface Types). Sie finden die WIT-Schemas im Azure IoT Operations-Beispiel-Repository.

Diese Bindungen stellen Folgendes bereit:

Typisierte Schnittstellen für Operatoren:

from map_impl import exports, imports
from map_impl.imports import types

# Implement the operator interface
class Map(exports.Map):
    def init(self, configuration) -> bool:
        # Access configuration parameters
        threshold = configuration.get_parameter("temperature_threshold")
        unit = configuration.get_parameter("output_unit", default="celsius")
        
        imports.logger.log(imports.logger.Level.INFO, "my-operator", 
                          f"Initialized with threshold={threshold}, unit={unit}")
        return True
    
    def process(self, message: types.DataModel) -> types.DataModel:
        # Your processing logic here
        return message

Ausführliche Informationen zum Definieren von Konfigurationsparametern in Diagrammdefinitionen finden Sie unter Modulkonfigurationsparameter.

Protokollierung über Importe:

# Access to structured logging
imports.logger.log(imports.logger.Level.INFO, "my-operator", "Processing started")
imports.logger.log(imports.logger.Level.ERROR, "my-operator", f"Error: {error}")

Fehlerbehandlung:

try:
    # Processing logic
    pass
except Exception as e:
    imports.logger.log(imports.logger.Level.ERROR, "my-operator", f"Error: {e}")
    return message  # Return original message on error