Tworzenie modułów WebAssembly (WASM) i definicji grafów dla wykresów przepływu danych

• Łańcuch narzędzi Rust udostępnia cargo, rustc oraz bibliotekę standardową wymaganą do kompilowania operatorów. Zainstaluj przy użyciu:

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

• Element docelowy WASM wasm32-wasip2 wymagany do kompilowania składników WASM operacji usługi Azure IoT. Dodaj za pomocą:

  rustup target add wasm32-wasip2
  

• Narzędzia kompilacji udostępniają funkcjonalności używane przez buildery i systemy CI do weryfikacji i pakietowania artefaktów WASM. Zainstaluj przy użyciu:

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

• Środowisko Python w wersji 3.8 lub nowszej

• componentize-py generuje powiązania i tworzy składniki WASM ze źródeł języka Python, które są zgodne ze schematami WIT operacji Azure IoT. Zainstaluj przy użyciu:

  pip install "componentize-py==0.14"
  

Dostęp do zestawu SDK WASM Rust można uzyskać za pomocą niestandardowego rejestru DevOps platformy Microsoft Azure. Zamiast używać zmiennych środowiskowych, skonfiguruj dostęp za pomocą pliku konfiguracji obszaru roboczego:

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

Ta konfiguracja odzwierciedla przykładowy układ pod adresem samples/wasm/.cargo/config.toml i przechowuje ustawienia rejestru w kontroli wersji.

Programowanie w języku Python używa składnika componentize-py z typami interfejsów WebAssembly (WIT) na potrzeby generowania kodu. Schematy funkcji WIT definiują interfejsy między kodem języka Python i środowiskiem uruchomieniowym WASM.

Pobierz schematy funkcji WIT: wymagane schematy można znaleźć w repozytorium przykładów operacji usługi Azure IoT. Sklonuj lub pobierz te schematy do środowiska deweloperskiego:

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

Schematy znajdują się w lokalizacji explore-iot-operations/samples/wasm-python/schema/ i zawierają definicje interfejsów dla wszystkich obsługiwanych typów operatorów, takich jak mapa, filtr i gałąź.

Nie potrzebujesz żadnej innej konfiguracji środowiska poza instalowaniem wymagań wstępnych i uzyskiwaniem schematów funkcji WIT.

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

Konfigurowanie pliku Cargo.toml

Edytuj plik, Cargo.toml aby uwzględnić zależności dla zestawu SDK WASM i innych bibliotek:

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

Uwagi dotyczące wersji i rejestru:

• Wersja zestawu SDK (=1.1.3) jest zgodna z aktualnymi przykładami; utrzymywanie jej w stałej wersji pozwala uniknąć zmian powodujących niezgodność.
• registry = "aio-wg" pasuje do wpisu rejestru zdefiniowanego w pliku .cargo/config.toml.

Objaśniono kluczowe zależności:

• wit-bindgen: Generuje wiązania Rust z definicji WebAssembly Interface Types (WIT), umożliwiając kodowi interakcję ze środowiskiem uruchomieniowym WASM.
• wasm_graph_sdk: Zestaw SDK operacji usługi Azure IoT udostępnia makra operatorów, takie jak #[map_operator] i #[filter_operator], oraz interfejsy API hosta na potrzeby rejestrowania, metryk i zarządzania stanem.
• serde + serde_json: biblioteki przetwarzania JSON do analizowania i generowania ładunków danych. default-features = false optymalizuje pod kątem ograniczeń rozmiaru WASM.
• crate-type = ["cdylib"]: kompiluje bibliotekę Rust jako bibliotekę dynamiczną zgodną z językiem C, która jest wymagana do generowania modułu WASM.

Utwórz plik w języku Python dla operatora. Nazwa pliku powinna być zgodna z nazwą modułu:

# Create your operator file
touch temperature_converter.py

Moduły WASM języka Python nie wymagają innych plików konfiguracji projektu. Klasa języka Python, która implementuje interfejs operatora, definiuje strukturę modułu.

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

Repozytorium przykładów operacji usługi Azure IoT obsługuje konstruktora Rust Docker i zawiera wszystkie niezbędne zależności. Aby uzyskać szczegółową dokumentację, zobacz Rust Docker builder usage (Użycie narzędzia 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

Opcje kompilacji platformy Docker:

• --app-name: Musi być zgodna z nazwą skrzyni rdzy z Cargo.toml
• --build-mode: wybierz release (ustawienie domyślne) dla zoptymalizowanych kompilacji lub debug kompilacji deweloperskich z symbolami

Repozytorium przykładów operacji usługi Azure IoT obsługuje konstruktora platformy Docker języka Python i zawiera wszystkie niezbędne zależności i schematy. Aby uzyskać szczegółową dokumentację, zobacz Użycie narzędzia Docker Builder w języku 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

Opcje kompilacji platformy Docker:

• --app-name: musi być zgodna .py z nazwą pliku języka Python bez rozszerzenia
• --app-type: Określ typ operatora (map, filter, branchitp.)
• --build-mode: wybierz release (ustawienie domyślne) dla zoptymalizowanych kompilacji lub debug kompilacji deweloperskich z symbolami

Aby zapoznać się z kompleksowymi przykładami, zobacz przykłady rust w repozytorium przykładów. Kompletne implementacje obejmują:

• Operatory mapy: przekształcanie i logika konwersji danych
• Operatory filtrów: warunkowe przetwarzanie i walidacja danych
• Operatory gałęzi: routing wielościeżkowy na podstawie zawartości danych
• Operatory kumulowania: agregacja okien czasowych i przetwarzanie statystyczne
• Operatory opóźnień: kontrolka przetwarzania na podstawie czasu

W przykładach pokazano działające implementacje, które pokazują pełną strukturę dla każdego typu operatora, w tym odpowiednie wzorce obsługi błędów i rejestrowania.

Aby zapoznać się z kompleksowymi przykładami , zobacz przykłady języka Python w repozytorium przykładów. Kompletne implementacje obejmują:

• Operatory mapy: przekształcanie i logika konwersji danych
• Operatory filtrów: warunkowe przetwarzanie i walidacja danych
• Operatory gałęzi: routing wielościeżkowy na podstawie zawartości danych

Zestaw SDK WASM Rust udostępnia kompleksowe narzędzia programistyczne:

Makra operatorów

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
}

Parametry konfiguracji modułu

Operatorzy WASM mogą odbierać parametry konfiguracji środowiska uruchomieniowego za pośrednictwem ModuleConfiguration struktury przekazanej init do funkcji. Te parametry definiuje się w definicji grafu, co umożliwia dostosowanie środowiska uruchomieniowego bez ponownego kompilowania modułów.

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
}

Aby uzyskać szczegółowe informacje na temat definiowania parametrów konfiguracji w definicjach grafu, zobacz Parametry konfiguracji modułu.

Interfejsy API hosta

Użyj zestawu SDK do pracy z usługami rozproszonymi:

Magazyn stanów dla danych trwałych:

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

Rejestrowanie strukturalne:

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

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

Metryki zgodne z technologią 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))?;

Tworzenie aplikacji WASM w języku Python nie korzysta z tradycyjnego zestawu SDK. Zamiast tego używasz wygenerowanych powiązań z typów interfejsów zestawu WebAssembly (WIT). Schematy funkcji WIT można znaleźć w repozytorium przykładów operacji usługi Azure IoT.

Te powiązania zapewniają:

Interfejsy typizowane dla operatorów:

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

Aby uzyskać szczegółowe informacje na temat definiowania parametrów konfiguracji w definicjach grafu, zobacz Parametry konfiguracji modułu.

Rejestrowanie za pośrednictwem importu:

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

Obsługa błędów:

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