Desenvolver módulos WebAssembly (WASM) e definições de gráficos para gráficos de fluxo de dados

• A cadeia de ferramentas Rust fornece cargo, rustc, e a biblioteca padrão necessária para compilar operadores. Instale com:

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

• É necessário o alvo WASM wasm32-wasip2 para construir os componentes WASM do Azure IoT Operations. Adicione com:

  rustup target add wasm32-wasip2
  

• As ferramentas de build fornecem utilitários usados pelos construtores e pela CI para validar e empacotar artefactos WASM. Instale com:

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

• Python 3.8 ou posterior

• componentize-py gera bindings e produz componentes WASM a partir de fontes Python que correspondem aos esquemas WIT do Azure IoT Operations. Instale com:

  pip install "componentize-py==0.14"
  

Pode aceder ao WASM Rust SDK através de um registo personalizado Microsoft Azure DevOps. Em vez de usar variáveis de ambiente, configure o acesso com um ficheiro de configuração de workspace:

# .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 configuração reflete o layout do exemplo em samples/wasm/.cargo/config.toml e mantém as configurações do registo no controlo de versões.

O desenvolvimento Python usa componentize-py com WebAssembly Interface Types (WIT) para geração de código. Os esquemas WIT definem as interfaces entre seu código Python e o tempo de execução do WASM.

Obtenha os esquemas WIT: Pode encontrar os esquemas necessários no repositório de exemplos Azure IoT Operations. Clone ou transfira estes esquemas para o seu ambiente de desenvolvimento:

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

Os esquemas estão localizados em explore-iot-operations/samples/wasm-python/schema/ e incluem definições de interface para todos os tipos de operadores suportados, como mapa, filtro e ramo.

Você não precisa de nenhuma outra configuração de ambiente além de instalar os pré-requisitos e obter os esquemas WIT.

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

Configurar o Cargo.toml

Edite o Cargo.toml arquivo para incluir dependências para o SDK do WASM e outras 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 versões e registo:

• A versão SDK (=1.1.3) está alinhada com as amostras atuais; mantê-la fixada evita alterações incompatíveis.
• registry = "aio-wg" corresponde à entrada do registo definida em .cargo/config.toml.

Principais dependências explicadas:

• wit-bindgen: Gera bindings Rust a partir das definições de Tipos de Interface WebAssembly (WIT), permitindo que o seu código interfaça com o runtime WASM.
• wasm_graph_sdk: Azure IoT Operations SDK que fornece macros de operador, como #[map_operator] e #[filter_operator], e APIs de host para registo, métricas e gestão de estados.
• serde + serde_json: Bibliotecas de processamento JSON para análise e geração de cargas úteis de dados. default-features = false otimiza para as restrições de tamanho do WASM.
• crate-type = ["cdylib"]: Compila a biblioteca Rust como uma biblioteca dinâmica compatível com C, necessária para a geração de módulos WASM.

Crie um arquivo Python para seu operador. O nome do arquivo deve corresponder ao nome do módulo:

# Create your operator file
touch temperature_converter.py

Os módulos Python WASM não requerem outros arquivos de configuração do projeto. A classe Python que implementa a interface do operador define a estrutura do 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

O repositório de exemplos Azure IoT Operations mantém o construtor Rust Docker e inclui todas as dependências necessárias. Para obter documentação detalhada, consulte Rust Docker builder usage.

# 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

Opções de compilação do Docker:

• --app-name: Deve corresponder ao nome da sua caixa Rust de Cargo.toml
• --build-mode: Escolha release (padrão) para compilações otimizadas ou debug para compilações de desenvolvimento com símbolos

O repositório de exemplos Azure IoT Operations mantém o construtor Python Docker e inclui todas as dependências e esquemas necessários. Para obter documentação detalhada, consulte Uso do construtor Python Docker.

# 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

Opções de compilação do Docker:

• --app-name: Deve corresponder ao seu nome de arquivo Python sem a .py extensão
• --app-type: Especifique o tipo de operador (map, filter, branch, etc.)
• --build-mode: Escolha release (padrão) para compilações otimizadas ou debug para compilações de desenvolvimento com símbolos

Para obter exemplos abrangentes, consulte os exemplos de Rust no repositório de amostras. As implementações completas incluem:

• Operadores de mapas: Lógica de transformação e conversão de dados
• Operadores de filtro: processamento e validação condicional de dados
• Operadores de filial: roteamento de vários caminhos com base no conteúdo de dados
• Acumular operadores: Agregação com janela de tempo e processamento estatístico
• Operadores de atraso: controle de processamento baseado no tempo

Os exemplos demonstram implementações de trabalho que mostram a estrutura completa para cada tipo de operador, incluindo o tratamento adequado de erros e padrões de registro.

Para obter exemplos abrangentes, consulte os exemplos Python no repositório de exemplos. As implementações completas incluem:

• Operadores de mapas: Lógica de transformação e conversão de dados
• Operadores de filtro: processamento e validação condicional de dados
• Operadores de filial: roteamento de vários caminhos com base no conteúdo de dados

O WASM Rust SDK fornece ferramentas de desenvolvimento abrangentes:

Macros do 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 configuração do módulo

Seus operadores WASM podem receber parâmetros de configuração de tempo de execução através da ModuleConfiguration struct passada para a init função. Define-se estes parâmetros na definição do grafo, que permite personalizar o tempo de execução sem ter de reconstruir 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 obter informações detalhadas sobre como definir parâmetros de configuração em definições de gráficos, consulte Parâmetros de configuração do módulo.

APIs de Hospedagem

Use o SDK para trabalhar com serviços distribuídos:

Armazenamento de estado para dados 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)?;

Registo estruturado:

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 compatíveis com 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))?;

O desenvolvimento Python WASM não usa um SDK tradicional. Em vez disso, você usa associações geradas de WebAssembly Interface Types (WIT). Pode encontrar os esquemas WIT no repositório de exemplos Azure IoT Operations.

Estas vinculações fornecem:

Interfaces digitadas 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 obter informações detalhadas sobre como definir parâmetros de configuração em definições de gráficos, consulte Parâmetros de configuração do módulo.

Registro em log através de importações:

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

Tratamento de erros:

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