Desarrollo de módulos webAssembly (WASM) y definiciones de grafos para gráficos de flujo de datos

• La cadena de herramientas de Rust proporciona cargo, rustcy la biblioteca estándar necesaria para compilar operadores. Instale con:

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

• Objetivo wasm32-wasip2 de WASM necesario para compilar componentes WASM de Operaciones de IoT de Azure. Agregue con:

  rustup target add wasm32-wasip2
  

• Las herramientas de compilación proporcionan utilidades que usan los compiladores y CI para validar y empaquetar artefactos WASM. Instale con:

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

• Python 3.8 o versiones posteriores

• componentize-py genera enlaces y genera componentes WASM de orígenes de Python que coinciden con los esquemas WIT de Azure IoT Operations. Instale con:

  pip install "componentize-py==0.14"
  

Puede acceder al SDK de WASM Rust a través de un registro personalizado de Microsoft Azure DevOps. En lugar de usar variables de entorno, configure el acceso con un archivo de configuración del área de trabajo:

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

Esta configuración replica el diseño de ejemplo en samples/wasm/.cargo/config.toml y mantiene la configuración del Registro dentro del control de versiones.

El desarrollo de Python usa componentize-py con tipos de interfaz WebAssembly (WIT) para la generación de código. Los esquemas WIT definen las interfaces entre el código de Python y el entorno de ejecución de WASM.

Obtenga los esquemas WIT: puede encontrar los esquemas necesarios en el repositorio de ejemplos de operaciones de Azure IoT. Clone o descargue estos esquemas en el entorno de desarrollo:

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

Los esquemas se encuentran en explore-iot-operations/samples/wasm-python/schema/ e incluyen definiciones de interfaz para todos los tipos de operador admitidos, como mapa, filtro y rama.

No necesita ninguna otra configuración de entorno más allá de instalar los requisitos previos y obtener los esquemas WIT.

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

Configuración de Cargo.toml

Edite el archivo Cargo.toml para incluir dependencias para el SDK de WASM y otras bibliotecas:

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

Notas sobre las versiones y el registro:

• La versión del SDK (=1.1.3) se alinea con los ejemplos actuales; mantenerla anclada evita cambios importantes.
• registry = "aio-wg" coincide con la entrada del Registro definida en .cargo/config.toml.

Dependencias clave explicadas:

• wit-bindgen: genera vinculaciones de Rust a partir de definiciones de tipos de interfaz de WebAssembly (WIT), permitiendo que el código interactúe con el entorno de ejecución WASM.
• wasm_graph_sdk: SDK de operaciones de Azure IoT que proporciona macros de operador, como #[map_operator] y #[filter_operator], y API de host para el registro, las métricas y la administración de estado.
• serde + serde_json: bibliotecas de procesamiento JSON para analizar y generar cargas de datos. default-features = false optimiza las restricciones de tamaño de WASM.
• crate-type = ["cdylib"]: compila la biblioteca de Rust como una biblioteca dinámica compatible con C, que es necesaria para la generación de módulos WASM.

Cree un archivo de Python para el operador. El nombre de archivo debe coincidir con el nombre del módulo:

# Create your operator file
touch temperature_converter.py

Los módulos WASM de Python no requieren otros archivos de configuración del proyecto. La clase de Python que implementa la interfaz de operador define la estructura del módulo.

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

El repositorio de ejemplos de Azure IoT Operations mantiene el generador de Docker de Rust e incluye todas las dependencias necesarias. Para obtener documentación detallada, consulte Uso del generador de Docker de Rust.

# 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

Opciones de compilación de Docker:

• --app-name: debe coincidir con el nombre de la caja de Rust desde Cargo.toml
• --build-mode: elija release (valor predeterminado) para compilaciones optimizadas o debug para compilaciones de desarrollo con símbolos

El repositorio de ejemplos de operaciones de IoT de Azure mantiene el generador de Docker de Python e incluye todas las dependencias y esquemas necesarios. Para obtener documentación detallada, consulte Uso del generador de Docker de Python.

# 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

Opciones de compilación de Docker:

• --app-name: debe coincidir con el nombre de archivo de Python sin la extensión .py
• --app-type: especifique el tipo de operador (map, filter, branch, etc.)
• --build-mode: elija release (valor predeterminado) para compilaciones optimizadas o debug para compilaciones de desarrollo con símbolos

Para obtener ejemplos completos, consulte los ejemplos de Rust en el repositorio de ejemplos. Las implementaciones completas incluyen:

• Operadores de mapa: lógica de transformación y conversión de datos
• Operadores de filtro: procesamiento y validación de datos condicionales
• Operadores de rama: enrutamiento de varias rutas en función del contenido de datos
• Operadores acumulados: agregación en período de tiempo y procesamiento estadístico
• Operadores de retraso: control de procesamiento basado en tiempo

Los ejemplos muestran implementaciones de trabajo que muestran la estructura completa de cada tipo de operador, incluidos los patrones de registro y control de errores adecuados.

Para obtener ejemplos completos, consulte los ejemplos de Python en el repositorio de ejemplos. Las implementaciones completas incluyen:

• Operadores de mapa: lógica de transformación y conversión de datos
• Operadores de filtro: procesamiento y validación de datos condicionales
• Operadores de rama: enrutamiento de varias rutas en función del contenido de datos

El SDK de WASM Rust proporciona herramientas de desarrollo completas:

Macros de operador

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
}

Parámetros de configuración del módulo

Los operadores WASM pueden recibir parámetros de configuración en tiempo de ejecución a través de la estructura ModuleConfiguration que se pasa a la función init. Estos parámetros se definen en la definición del grafo, lo que le permite personalizar el entorno de ejecución sin volver a generar módulos.

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
}

Para obtener información detallada sobre cómo definir parámetros de configuración en definiciones de grafos, consulte Parámetros de configuración del módulo.

API de host

Use el SDK para trabajar con servicios distribuidos:

Almacén de estado para datos persistentes:

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)?;

Registro estructurado:

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

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

Métricas compatibles con OpenTelemetry:

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))?;

El desarrollo de WASM de Python no usa un SDK tradicional. En su lugar, se usan enlaces generados de tipos de interfaz WebAssembly (WIT). Puede encontrar los esquemas WIT en el repositorio de ejemplos de operaciones de Azure IoT.

Estas vinculaciones proporcionan:

Interfaces con tipo para operadores:

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

Para obtener información detallada sobre cómo definir parámetros de configuración en definiciones de grafos, consulte Parámetros de configuración del módulo.

Registro mediante importaciones:

# 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}")

Control de errores:

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