Desenvolver módulos de WebAssembly (WASM) e definições de grafo para grafos 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
  

• Destino WASM wasm32-wasip2 necessário para criar componentes WASM das Operações do Azure IoT. Adicione com:

  rustup target add wasm32-wasip2
  

• As ferramentas de compilação fornecem utilitários usados pelos compiladores e pela integração contínua (CI) para validar e empacotar artefatos WASM. Instale com:

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

• Python 3.8 ou posterior

• componentize-py gera associações e produz componentes WASM de fontes do Python que correspondem aos esquemas WIT das Operações IoT do Azure. Instale com:

  pip install "componentize-py==0.14"
  

Você pode acessar o SDK do RUST do WASM por meio de um registro personalizado do Microsoft Azure DevOps. Em vez de usar variáveis de ambiente, configure o acesso com um arquivo de configuração do 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

Essa configuração espelha o layout de exemplo em samples/wasm/.cargo/config.toml e mantém as configurações do Registro sob controle de versão.

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

Obtenha os esquemas wit: você pode encontrar os esquemas necessários no repositório de exemplos de Operações IoT do Azure. Clone ou baixe esses esquemas em 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 com suporte, como mapa, filtro e ramificação.

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

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

Configurar Cargo.toml

Edite o arquivo Cargo.toml 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"]

Anotações sobre versões e registro:

• A versão do SDK (=1.1.3) se alinha com os exemplos atuais; mantê-la fixada evita alterações significativas.
• registry = "aio-wg" corresponde à entrada do Registro definida em .cargo/config.toml.

As principais dependências explicadas:

• wit-bindgen: gera associações Rust de definições WIT (Tipos de Interface WebAssembly), permitindo que seu código interaja com o runtime WASM.
• wasm_graph_sdk: SDK de Operações de IoT do Azure fornecendo macros de operador, como #[map_operator] e #[filter_operator], e APIs de host para registro em log, métricas e gerenciamento de estado.
• serde + serde_json: bibliotecas de processamento JSON para analisar e gerar cargas de dados. default-features = false otimiza para restrições de tamanho do WASM.
• crate-type = ["cdylib"]: compila a biblioteca Rust como uma biblioteca dinâmica compatível com C, que é necessária para a geração do módulo WASM.

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

# Create your operator file
touch temperature_converter.py

Os módulos WASM do Python não exigem outros arquivos de configuração de 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 de Operações IoT do Azure mantém o construtor do Rust Docker e inclui todas as dependências necessárias. Para obter uma documentação detalhada, veja Uso do Construtor do Docker para 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

Opções de build do Docker:

• --app-name: deve corresponder ao seu nome de crate do Rust de Cargo.toml
• --build-mode: escolha release (padrão) para builds otimizados ou debug para builds de desenvolvimento com símbolos

O repositório de exemplos de Operações IoT do Azure mantém o construtor do Docker do Python e inclui todas as dependências e esquemas necessários. Para obter uma documentação detalhada, veja Uso do construtor do Docker para 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

Opções de build do Docker:

• --app-name: deve corresponder ao nome do arquivo Python sem a extensão .py
• --app-type: especifique o tipo de operador (map, filter, branch etc.)
• --build-mode: escolha release (padrão) para builds otimizados ou debug para builds de desenvolvimento com símbolos

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

• Operadores de mapa: transformação de dados e lógica de conversão
• Operadores de filtro: processamento e validação de dados condicionais
• Operadores de ramificação: roteamento de vários caminhos com base no conteúdo de dados
• Operadores acumulados: agregação em janelas de tempo e processamento estatístico
• Operadores de atraso: controle de processamento baseado em tempo

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

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

• Operadores de mapa: transformação de dados e lógica de conversão
• Operadores de filtro: processamento e validação de dados condicionais
• Operadores de ramificação: roteamento de vários caminhos com base no conteúdo de dados

O SDK do WASM Rust fornece ferramentas de desenvolvimento abrangentes:

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

Seus operadores WASM podem receber parâmetros de configuração de runtime por meio do struct ModuleConfiguration passado para a função init. Você define esses parâmetros na definição de grafo, que permite personalizar o runtime sem recompilar 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 os parâmetros de configuração em definições de grafo, veja Parâmetros de configuração do módulo.

APIs de Host

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

Repositório 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)?;

Registro em log 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 de WASM do Python não usa um SDK tradicional. Você usa associações geradas de WIT (Tipos de Interface WebAssembly). Você pode encontrar os esquemas WIT no repositório de exemplos de Operações IoT do Azure.

Essas associações fornecem:

Interfaces tipada 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 os parâmetros de configuração em definições de grafo, veja Parâmetros de configuração do módulo.

Registrar em log por meio 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