Używanie zestawu WebAssembly (WASM) z wykresami przepływu danych

Wykresy przepływu danych operacji usługi Azure IoT obsługują moduły WebAssembly (WASM) na potrzeby niestandardowego przetwarzania danych na urządzeniach brzegowych. Możesz wdrożyć niestandardową logikę biznesową i przekształcenia danych w ramach potoków przepływu danych.

Wymagania wstępne

• Wdróż instancję operacyjną Azure IoT w klastrze Kubernetes z włączoną obsługą Arc. Aby uzyskać więcej informacji, zobacz Wdrażanie operacji usługi Azure IoT.
• Skonfiguruj rejestr kontenerów i dodaj przykładowe definicje grafu i moduły WASM, postępując zgodnie ze wskazówkami w temacie Wdrażanie modułów WebAssembly (WASM) i definicji grafu.

Przegląd

Moduły WebAssembly (WASM) w wykresach przepływu danych operacji usługi Azure IoT umożliwiają przetwarzanie danych na brzegu z wysoką wydajnością i zabezpieczeniami. Program WASM działa w środowisku w trybie piaskownicy i obsługuje język Rust i Python.

Jak działają wykresy przepływu danych WASM

Implementacja przepływu danych WASM jest zgodna z tym przepływem pracy:

1. Opracowywanie modułów WASM: pisanie niestandardowej logiki przetwarzania w obsługiwanym języku i kompilowanie jej do formatu modelu składników zestawu WebAssembly. Aby dowiedzieć się więcej, zobacz:
   • Tworzenie modułów WASM dla przepływów danych w programie VS Code
   • Tworzenie modułów zestawu WebAssembly (WASM)
2. Tworzenie definicji grafu: definiowanie sposobu, w jaki dane przechodzą przez moduły przy użyciu plików konfiguracji YAML. Aby dowiedzieć się więcej, zobacz Konfigurowanie definicji grafu zestawu WebAssembly.
3. Przechowywanie artefaktów w rejestrze: wypychanie skompilowanych modułów WASM i definicji grafów do rejestru kontenerów przy użyciu narzędzi zgodnych z protokołem OCI, takich jak ORAS. Aby dowiedzieć się więcej, zobacz Deploy WebAssembly (WASM) modules and graph definitions (Wdrażanie modułów WebAssembly (WASM) i definicji grafu.
4. Konfigurowanie punktów końcowych rejestru: skonfiguruj szczegóły uwierzytelniania i połączenia, aby operacje usługi Azure IoT mogły uzyskiwać dostęp do rejestru kontenerów. Aby dowiedzieć się więcej, zobacz Konfigurowanie punktów końcowych rejestru.
5. Tworzenie grafu przepływu danych: użyj internetowego interfejsu użytkownika środowiska operacji lub plików Bicep, aby zdefiniować przepływ danych korzystający z definicji grafu.
6. Wdrażanie i wykonywanie: Operacje usługi Azure IoT ściąga definicje grafu i moduły WASM z rejestru kontenerów i uruchamia je.

W poniższych przykładach pokazano, jak skonfigurować wykresy przepływu danych WASM dla typowych scenariuszy. W przykładach są używane wartości zakodowane na stałe i uproszczone konfiguracje, dzięki czemu można szybko rozpocząć pracę.

Przykład 1. Podstawowe wdrożenie z jednym modułem WASM

W tym przykładzie dane temperatury są konwertowane z fahrenheita na stopnie Celsjusza przy użyciu modułu WASM. Kod źródłowy modułu temperature jest dostępny w usłudze GitHub. Jeśli wykonano przykładowe kroki opisane w artykule Wdrażanie modułów WebAssembly (WASM) i definicji grafu, graph-simple:1.0.0 definicja grafu i wstępnie skompilowany temperature:1.0.0 moduł znajdują się już w rejestrze kontenerów.

Jak to działa

Definicja grafu tworzy prosty, trzyetapowy potok:

1. Źródło: Odbiera dane temperatury z MQTT
2. Mapa: przetwarza dane za pomocą modułu WASM temperatury
3. Ujście: wysyła przekonwertowane dane z powrotem do MQTT

Aby dowiedzieć się więcej o sposobie działania prostej definicji grafu i jej strukturze, zobacz Przykład 1: Prosta definicja grafu.

Format danych wejściowych:

{"temperature": {"value": 100.0, "unit": "F"}}

Format danych wyjściowych:

{"temperature": {"value": 37.8, "unit": "C"}}

Poniższa konfiguracja tworzy graf przepływu danych, który używa tego potoku konwersji temperatury. Wykres przepływu danych odwołuje się do graph-simple:1.0.0 definicji grafu YAML i ściąga moduł temperatury z rejestru kontenerów.

Konfigurowanie grafu przepływu danych

Ta konfiguracja definiuje trzy węzły, które implementują przepływ pracy konwersji temperatury: węzeł źródłowy, który subskrybuje przychodzące dane temperatury, węzeł przetwarzania grafu z uruchomionym modułem WASM i węzeł docelowy, który publikuje przekonwertowane wyniki.

Zasób grafu przepływu danych opakowuje artefakt definiujący graf i łączy swoje abstrakcyjne operacje źródła/odbiornika z specyficznymi punktami końcowymi.

• Operacja definicji source grafu łączy się z węzłem źródłowym przepływu danych (temat MQTT)
• Operacja definicji sink grafu łączy się z węzłem docelowym przepływu danych (temat MQTT)
• Operacje przetwarzania definicji grafu są uruchamiane w węźle przetwarzania grafu

Ta separacja umożliwia wdrożenie tej samej definicji grafu z różnymi punktami końcowymi w różnych środowiskach przy zachowaniu logiki przetwarzania bez zmian.

param aioInstanceName string = '<AIO_INSTANCE_NAME>'
param customLocationName string = '<CUSTOM_LOCATION_NAME>'
param dataflowGraphName string = '<GRAPH_NAME>'
param registryEndpointName string = '<REGISTRY_ENDPOINT_NAME>'

resource aioInstance 'Microsoft.IoTOperations/instances@2025-10-01' existing = {
  name: aioInstanceName
}

resource customLocation 'Microsoft.ExtendedLocation/customLocations@2021-08-31-15' existing = {
  name: customLocationName
}

resource defaultDataflowProfile 'Microsoft.IoTOperations/instances/dataflowProfiles@2025-10-01' existing = {
  parent: aioInstance
  name: 'default'
}

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2025-10-01' = {
  parent: defaultDataflowProfile
  name: dataflowGraphName
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    mode: 'Enabled'
    requestDiskPersistence: 'Disabled'
    nodes: [
      {
        nodeType: 'Source'
        name: 'temperature-source'
        sourceSettings: {
          endpointRef: 'default'
          dataSources: [
            'sensor/temperature/raw'
          ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'temperature-processor'
        graphSettings: {
          registryEndpointRef: registryEndpointName
          artifact: 'graph-simple:1.0.0'
          configuration: [
            {
              key: 'key1'
              value: 'example-value'
            }
          ]
        }
      }
      {
        nodeType: 'Destination'
        name: 'temperature-destination'
        destinationSettings: {
          endpointRef: 'default'
          dataDestination: 'sensor/temperature/processed'
        }
      }
    ]
    nodeConnections: [
      {
        from: {
          name: 'temperature-source'
        }
        to: {
          name: 'temperature-processor'
        }
      }
      {
        from: {
          name: 'temperature-processor'
        }
        to: {
          name: 'temperature-destination'
        }
      }
    ]
  }
}

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
  name: <GRAPH_NAME>
  namespace: azure-iot-operations
spec:
  requestDiskPersistence: Disabled
  profileRef: default

  nodes:
    - nodeType: Source
      name: temperature-source
      sourceSettings:
        endpointRef: default
        dataSources:
          - sensor/temperature/raw

    - nodeType: Graph
      name: temperature-processor
      graphSettings:
        registryEndpointRef: <REGISTRY_ENDPOINT_NAME>
        artifact: graph-simple:1.0.0
        configuration:
          - key: key1
            value: example-value

    - nodeType: Destination
      name: temperature-destination
      destinationSettings:
        endpointRef: default
        dataDestination: sensor/temperature/processed

  nodeConnections:
    - from:
        name: temperature-source
      to:
        name: temperature-processor

    - from:
        name: temperature-processor
      to:
        name: temperature-destination

kubectl apply -f dataflow-graph.yaml

Testowanie przepływu danych

Aby przetestować przepływ danych, wyślij komunikaty MQTT z klastra. Najpierw wdróż zasobnik klienta MQTT, postępując zgodnie z instrukcjami w temacie Testowanie łączności z brokerem MQTT przy użyciu klientów MQTT. Klient MQTT udostępnia tokeny uwierzytelniania i certyfikaty do nawiązywania połączenia z brokerem. Aby wdrożyć klienta MQTT, uruchom następujące polecenie:

kubectl apply -f https://raw.githubusercontent.com/Azure-Samples/explore-iot-operations/main/samples/quickstarts/mqtt-client.yaml

Wysyłanie komunikatów o temperaturze

W pierwszej sesji terminalu utwórz i uruchom skrypt w celu wysyłania danych temperatury w fahrenheit:

# Connect to the MQTT client pod
kubectl exec --stdin --tty mqtt-client -n azure-iot-operations -- sh -c '
# Create and run temperature.sh from within the MQTT client pod
while true; do
  # Generate a random temperature value between 0 and 6000 Fahrenheit
  random_value=$(shuf -i 0-6000 -n 1)
  payload="{\"temperature\":{\"value\":$random_value,\"unit\":\"F\"}}"

  echo "Publishing temperature: $payload"

  # Publish to the input topic
  mosquitto_pub -h aio-broker -p 18883 \
    -m "$payload" \
    -t "sensor/temperature/raw" \
    -d \
    --cafile /var/run/certs/ca.crt \
    -D PUBLISH user-property __ts $(date +%s)000:0:df \
    -D CONNECT authentication-method 'K8S-SAT' \
    -D CONNECT authentication-data $(cat /var/run/secrets/tokens/broker-sat)

  sleep 1
done'

Skrypt publikuje dane losowej temperatury w sensor/temperature/raw temacie co sekundę. Powinien wyglądać następująco:

Publishing temperature: {"temperature":{"value":1234,"unit":"F"}}
Publishing temperature: {"temperature":{"value":5678,"unit":"F"}}

Pozostaw skrypt uruchomiony, aby kontynuować publikowanie danych dotyczących temperatury.

Subskrybowanie przetworzonych komunikatów

W drugiej sesji terminalu (również podłączonej do zasobnika klienta MQTT) zasubskrybuj temat wyjściowy, aby wyświetlić przekonwertowane wartości temperatury:

# Connect to the MQTT client pod
kubectl exec --stdin --tty mqtt-client -n azure-iot-operations -- sh -c '
mosquitto_sub -h aio-broker -p 18883 -t "sensor/temperature/processed" --cafile /var/run/certs/ca.crt \
-D CONNECT authentication-method "K8S-SAT" \
-D CONNECT authentication-data "$(cat /var/run/secrets/tokens/broker-sat)"'

Zobaczysz dane dotyczące temperatury przekonwertowane z fahrenheita na stopnie Celsjusza przez moduł WASM.

{"temperature":{"value":1292.2222222222222,"count":0,"max":0.0,"min":0.0,"average":0.0,"last":0.0,"unit":"C","overtemp":false}}
{"temperature":{"value":203.33333333333334,"count":0,"max":0.0,"min":0.0,"average":0.0,"last":0.0,"unit":"C","overtemp":false}}

Przykład 2. Wdrażanie złożonego grafu

W tym przykładzie przedstawiono zaawansowany przepływ pracy przetwarzania danych, który obsługuje wiele typów danych, takich jak temperatura, wilgotność i dane obrazu. Złożona definicja grafu organizuje wiele modułów WASM w celu przeprowadzania zaawansowanej analizy i wykrywania obiektów.

Jak to działa

Złożony graf przetwarza trzy strumienie danych i łączy je w wzbogaconą analizę czujników:

• Przetwarzanie temperatury: konwertuje fahrenheita na stopnie Celsjusza, filtruje nieprawidłowe odczyty i oblicza statystyki
• Przetwarzanie wilgotności: Gromadzi pomiary wilgotności w odstępach czasu
• Przetwarzanie obrazów: wykonuje wykrywanie obiektów na migawkach aparatu i formatuje wyniki

Aby dowiedzieć się więcej o sposobie działania złożonej definicji grafu, jej strukturze i przepływie danych przez wiele etapów przetwarzania, zobacz Przykład 2: złożona definicja grafu.

Wykres używa wyspecjalizowanych modułów z kolekcji operatorów Rust.

Konfigurowanie złożonego grafu przepływu danych

Ta konfiguracja implementuje przepływ pracy przetwarzania wielu czujników przy użyciu graph-complex:1.0.0 definicji grafu YAML. Zwróć uwagę, że wdrożenie grafu przepływu danych jest podobne do przykładu Przykład 1 — oba używają tego samego wzorca z trzema węzłami (źródła, procesora grafu, miejsca docelowego), mimo że logika przetwarzania jest inna.

To podobieństwo występuje, ponieważ zasób grafu przepływu danych działa jako środowisko hosta, które ładuje i wykonuje definicje grafu. Rzeczywista logika przetwarzania znajduje się w definicji grafu (graph-simple:1.0.0 lub graph-complex:1.0.0), która zawiera specyfikację YAML operacji i połączeń między modułami WASM. Zasób grafu przepływu danych udostępnia infrastrukturę środowiska uruchomieniowego do ściągania definicji grafu, tworzenia wystąpień modułów i kierowania danych za pośrednictwem zdefiniowanego przepływu pracy.

param aioInstanceName string = '<AIO_INSTANCE_NAME>'
param customLocationName string = '<CUSTOM_LOCATION_NAME>'
param dataflowGraphName string = '<COMPLEX_GRAPH_NAME>'
param registryEndpointName string = '<REGISTRY_ENDPOINT_NAME>'

resource aioInstance 'Microsoft.IoTOperations/instances@2025-10-01' existing = {
  name: aioInstanceName
}

resource customLocation 'Microsoft.ExtendedLocation/customLocations@2021-08-15' existing = {
  name: customLocationName
}

resource defaultDataflowProfile 'Microsoft.IoTOperations/instances/dataflowProfiles@2025-10-01' existing = {
  parent: aioInstance
  name: 'default'
}

resource complexDataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2025-10-01' = {
  parent: defaultDataflowProfile
  name: dataflowGraphName
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    mode: 'Enabled'
    requestDiskPersistence: 'Disabled'
    nodes: [
      {
        nodeType: 'Source'
        name: 'sensor-source'
        sourceSettings: {
          endpointRef: 'default'
          dataSources: [
            'sensor/temperature/raw'
            'sensor/humidity/raw'
            'sensor/images/raw'
          ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'sensor-processor'
        graphSettings: {
          registryEndpointRef: registryEndpointName
          artifact: 'graph-complex:1.0.0'
          configuration: [
            {
              key: 'snapshot_topic'
              value: 'sensor/images/raw'
            }
            {
              key: 'key1'
              value: 'example-value'
            }
          ]
        }
      }
      {
        nodeType: 'Destination'
        name: 'analytics-destination'
        destinationSettings: {
          endpointRef: 'default'
          dataDestination: 'analytics/sensor/processed'
        }
      }
    ]
    nodeConnections: [
      {
        from: {
          name: 'sensor-source'
        }
        to: {
          name: 'sensor-processor'
        }
      }
      {
        from: {
          name: 'sensor-processor'
        }
        to: {
          name: 'analytics-destination'
        }
      }
    ]
  }
}

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
  name: <COMPLEX_GRAPH_NAME>
  namespace: azure-iot-operations
spec:
  requestDiskPersistence: Disabled
  profileRef: default

  nodes:
    - nodeType: Source
      name: sensor-source
      sourceSettings:
        endpointRef: default
        dataSources:
          - sensor/temperature/raw
          - sensor/humidity/raw
          - sensor/images/raw

    - nodeType: Graph
      name: sensor-processor
      graphSettings:
        registryEndpointRef: <REGISTRY_ENDPOINT_NAME>
        artifact: graph-complex:1.0.0
        configuration:
          - key: snapshot_topic
            value: sensor/images/raw
          - key: key1
            value: example-value

    - nodeType: Destination
      name: analytics-destination
      destinationSettings:
        endpointRef: default
        dataDestination: analytics/sensor/processed

  nodeConnections:
    - from:
        name: sensor-source
      to:
        name: sensor-processor

    - from:
        name: sensor-processor
      to:
        name: analytics-destination

Testowanie złożonego przepływu danych

Zanim zobaczysz jakiekolwiek dane wyjściowe, skonfiguruj dane źródłowe.

Przekazywanie plików obrazów RAW do zasobnika mqtt-client

Pliki obrazów są przeznaczone dla modułu snapshot do wykrywania obiektów na obrazach. Znajdują się one w folderze images w usłudze GitHub.

Najpierw sklonuj repozytorium, aby uzyskać dostęp do plików obrazów:

git clone https://github.com/Azure-Samples/explore-iot-operations.git
cd explore-iot-operations

Aby przekazać pliki obrazów RAW z ./samples/wasm/images folderu do mqtt-client zasobnika, możesz użyć następującego polecenia:

kubectl cp ./samples/wasm/images azure-iot-operations/mqtt-client:/tmp

Sprawdź, czy pliki są przekazywane:

kubectl exec -it mqtt-client -n azure-iot-operations -- ls /tmp/images

Powinna zostać wyświetlona lista plików w folderze /tmp/images .

beaker.raw          laptop.raw          sunny2.raw
binoculars.raw      lawnmower.raw       sunny4.raw
broom.raw           milkcan.raw         thimble.raw
camera.raw          photocopier.raw     tripod.raw
computer_mouse.raw  radiator.raw        typewriter.raw
daisy3.raw          screwdriver.raw     vacuum_cleaner.raw
digital_clock.raw   sewing_machine.raw
hammer.raw          sliding_door.raw

Publikowanie symulowanej temperatury, danych wilgotności i wysyłanie obrazów

Możesz połączyć polecenia służące do publikowania danych dotyczących temperatury, wilgotności i wysyłania obrazów do jednego skryptu. Użyj następującego polecenia:

# Connect to the MQTT client pod and run the script
kubectl exec --stdin --tty mqtt-client -n azure-iot-operations -- sh -c '
while true; do 
  # Generate a random temperature value between 0 and 6000
  temp_value=$(shuf -i 0-6000 -n 1)
  temp_payload="{\"temperature\":{\"value\":$temp_value,\"unit\":\"F\"}}"
  echo "Publishing temperature: $temp_payload"
  mosquitto_pub -h aio-broker -p 18883 \
    -m "$temp_payload" \
    -t "sensor/temperature/raw" \
    --cafile /var/run/certs/ca.crt \
    -D CONNECT authentication-method "K8S-SAT" \
    -D CONNECT authentication-data "$(cat /var/run/secrets/tokens/broker-sat)" \
    -D PUBLISH user-property __ts $(date +%s)000:0:df

  # Generate a random humidity value between 30 and 90
  humidity_value=$(shuf -i 30-90 -n 1)
  humidity_payload="{\"humidity\":{\"value\":$humidity_value}}"
  echo "Publishing humidity: $humidity_payload"
  mosquitto_pub -h aio-broker -p 18883 \
    -m "$humidity_payload" \
    -t "sensor/humidity/raw" \
    --cafile /var/run/certs/ca.crt \
    -D CONNECT authentication-method "K8S-SAT" \
    -D CONNECT authentication-data "$(cat /var/run/secrets/tokens/broker-sat)" \
    -D PUBLISH user-property __ts $(date +%s)000:0:df

  # Send an image every 2 seconds
  if [ $(( $(date +%s) % 2 )) -eq 0 ]; then
    file=$(ls /tmp/images/*.raw | shuf -n 1)
    echo "Sending file: $file"
    mosquitto_pub -h aio-broker -p 18883 \
      -f $file \
      -t "sensor/images/raw" \
      --cafile /var/run/certs/ca.crt \
      -D CONNECT authentication-method "K8S-SAT" \
      -D CONNECT authentication-data "$(cat /var/run/secrets/tokens/broker-sat)" \
      -D PUBLISH user-property __ts $(date +%s)000:0:df
  fi

  # Wait for 1 second before the next iteration
  sleep 1
done'

Sprawdzanie danych wyjściowych

W nowym terminalu zasubskrybuj temat wyjściowy:

kubectl exec --stdin --tty mqtt-client -n azure-iot-operations -- sh -c '
mosquitto_sub -h aio-broker -p 18883 -t "analytics/sensor/processed" --cafile /var/run/certs/ca.crt \
-D CONNECT authentication-method "K8S-SAT" \
-D CONNECT authentication-data "$(cat /var/run/secrets/tokens/broker-sat)"'

Dane wyjściowe wyglądają jak w poniższym przykładzie:

{"temperature":[{"count":9,"max":2984.4444444444443,"min":248.33333333333337,"average":1849.6296296296296,"last":2612.222222222222,"unit":"C","overtemp":true}],"humidity":[{"count":10,"max":76.0,"min":30.0,"average":49.7,"last":38.0}],"object":[{"result":"milk can; broom; screwdriver; binoculars, field glasses, opera glasses; toy terrier"}]}
{"temperature":[{"count":10,"max":2490.5555555555557,"min":430.55555555555554,"average":1442.6666666666667,"last":1270.5555555555557,"unit":"C","overtemp":true}],"humidity":[{"count":9,"max":87.0,"min":34.0,"average":57.666666666666664,"last":42.0}],"object":[{"result":"broom; Saint Bernard, St Bernard; radiator"}]}

W tym miejscu dane wyjściowe zawierają dane dotyczące temperatury i wilgotności, a także wykryte obiekty na obrazach.

Konfiguracja niestandardowych wykresów przepływu danych

Ta sekcja zawiera szczegółowe informacje na temat konfigurowania wykresów przepływu danych przy użyciu modułów WASM. Obejmuje wszystkie opcje konfiguracji, punkty końcowe przepływu danych i ustawienia zaawansowane.

Omówienie wykresu przepływu danych

Wykres przepływu danych definiuje sposób przepływu danych za pośrednictwem modułów WebAssembly na potrzeby przetwarzania. Każdy graf składa się z:

• Tryb, który kontroluje, czy wykres jest włączony, czy wyłączony
• Odwołanie do profilu łączącego się z profilem przepływu danych definiującym skalowanie i ustawienia zasobów
• Trwałość dysku, która opcjonalnie włącza trwały magazyn dla stanu grafu
• Węzły definiujące składniki źródłowe, obliczeniowe i docelowe
• Połączenia węzłów określające sposób przepływu danych między węzłami

Konfiguracja trybu

Właściwość mode określa, czy wykres przepływu danych aktywnie przetwarza dane. Możesz ustawić tryb na Enabled lub Disabled (bez uwzględniania wielkości liter). Po wyłączeniu wykres zatrzymuje przetwarzanie danych, ale zachowuje konfigurację.

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2025-10-01' = {
  // ... other properties
  properties: {
    mode: 'Enabled'  // or 'Disabled'
    // ... other properties
  }
}

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
  name: my-dataflow-graph
  namespace: azure-iot-operations
spec:
  mode: Enabled  # or Disabled
  # ... other properties

Dokumentacja profilu

Dokumentacja profilu łączy wykres przepływu danych z profilem przepływu danych, który definiuje ustawienia skalowania, liczby wystąpień i limity zasobów. Jeśli nie określisz odwołania do profilu, musisz zamiast tego użyć odwołania właściciela platformy Kubernetes. Większość scenariuszy używa domyślnego profilu udostępnianego przez operacje usługi Azure IoT.

resource defaultDataflowProfile 'Microsoft.IoTOperations/instances/dataflowProfiles@2025-10-01' existing = {
  parent: aioInstance
  name: 'default'
}

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2025-10-01' = {
  parent: defaultDataflowProfile  // This establishes the profile relationship
  // ... other properties
}

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
  name: my-dataflow-graph
  namespace: azure-iot-operations
spec:
  profileRef: default  # Reference to the profile name
  # ... other properties

Żądanie trwałości dysku

Żądanie trwałości dysku umożliwia grafom przepływu danych zachowanie stanu po ponownym uruchomieniu. Po włączeniu tej funkcji graf może odzyskać stan przetwarzania po ponownym uruchomieniu połączonego brokera. Ta funkcja jest przydatna w scenariuszach przetwarzania stanowego, w których utrata danych pośrednich byłaby problematyczna. Po włączeniu trwałości dysku żądania broker utrwala dane MQTT, takie jak komunikaty w kolejce subskrybentów, na dysku. Takie podejście zapewnia, że źródło danych przepływu danych nie spowoduje utraty danych podczas przestojów zasilania ani ponownego uruchomienia brokera. Broker utrzymuje optymalną wydajność, ponieważ trwałość jest skonfigurowana dla przepływu danych, więc tylko przepływy danych, które wymagają trwałości, używają tej funkcji.

Wykres przepływu danych sprawia, że to żądanie trwałości podczas subskrypcji przy użyciu właściwości użytkownika MQTTv5. Ta funkcja działa tylko wtedy, gdy:

• Przepływ danych używa brokera MQTT jako źródła (węzeł źródłowy z punktem końcowym MQTT)
• Broker MQTT ma włączoną trwałość Enabled z dynamicznym trybem trwałości ustawionym dla typu danych, takiego jak kolejki subskrybentów

Ta konfiguracja umożliwia klientom MQTT, takim jak wykresy przepływu danych, żądania trwałości dysku dla swoich subskrypcji przy użyciu właściwości użytkownika MQTTv5. Aby uzyskać szczegółową konfigurację trwałości brokera MQTT, zobacz Konfigurowanie trwałości brokera MQTT.

Ustawienie akceptuje Enabled wartość lub Disabled, z wartością Disabled domyślną.

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2025-10-01' = {
  // ... other properties
  properties: {
    requestDiskPersistence: 'Enabled'
    // ... other properties
  }
}

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
  name: my-dataflow-graph
  namespace: azure-iot-operations
spec:
  requestDiskPersistence: Enabled
  # ... other properties

Konfiguracja węzła

Węzły to bloki konstrukcyjne grafu przepływu danych. Każdy węzeł ma unikatową nazwę w grafie i wykonuje określoną funkcję. Istnieją trzy typy węzłów:

Węzły źródłowe

Węzły źródłowe definiują miejsce wprowadzania danych do grafu. Łączą się z punktami końcowymi przepływu danych, które odbierają dane z brokerów MQTT lub tematów platformy Kafka. Każdy węzeł źródłowy musi określać:

• Odwołanie do punktu końcowego wskazującego skonfigurowany punkt końcowy przepływu danych
• Źródła danych jako lista tematów MQTT lub tematów platformy Kafka do subskrybowania
• Dokumentacja zasobu (opcjonalnie) łącząca się z zasobem usługi Azure Device Registry na potrzeby wnioskowania schematu

Tablica źródeł danych umożliwia subskrybowanie wielu tematów bez modyfikowania konfiguracji punktu końcowego. Ta elastyczność umożliwia ponowne użycie punktu końcowego w różnych przepływach danych.

{
  nodeType: 'Source'
  name: 'sensor-data-source'
  sourceSettings: {
    endpointRef: 'default'  // References a data flow endpoint
    dataSources: [
      'sensor/+/temperature'  // MQTT topic with wildcard
      'sensor/+/humidity'
    ]
    assetRef: 'weather-station-asset'  // Optional asset reference
  }
}

- nodeType: Source
  name: sensor-data-source
  sourceSettings:
    endpointRef: default  # References a data flow endpoint
    dataSources:
      - sensor/+/temperature  # MQTT topic with wildcard
      - sensor/+/humidity
    assetRef: weather-station-asset  # Optional asset reference

Węzły przetwarzania grafu

Węzły przetwarzania grafu zawierają moduły WebAssembly, które przekształcają dane. Te węzły ściągają artefakty WASM z rejestrów kontenerów i wykonują je przy użyciu określonych parametrów konfiguracji. Każdy węzeł grafu wymaga:

• Odwołanie do punktu końcowego rejestru wskazującego punkt końcowy rejestru na potrzeby ściągania artefaktów
• Specyfikacja artefaktu definiująca nazwę i wersję modułu do ściągnięcia
• Parametry konfiguracji jako pary klucz-wartość przekazywane do modułu WASM

Tablica konfiguracji umożliwia dostosowanie zachowania modułu bez ponownego kompilowania artefaktu WASM. Typowe opcje konfiguracji obejmują parametry przetwarzania, progi, ustawienia konwersji i flagi funkcji.

{
  nodeType: 'Graph'
  name: 'temperature-processor'
  graphSettings: {
    registryEndpointRef: 'my-acr-endpoint'
    artifact: 'temperature-converter:2.1.0'
    configuration: [
      {
        key: 'input-unit'
        value: 'fahrenheit'
      }
      {
        key: 'output-unit' 
        value: 'celsius'
      }
      {
        key: 'precision'
        value: '2'
      }
      {
        key: 'enable-filtering'
        value: 'true'
      }
    ]
  }
}

- nodeType: Graph
  name: temperature-processor
  graphSettings:
    registryEndpointRef: my-acr-endpoint
    artifact: temperature-converter:2.1.0
    configuration:
      - key: input-unit
        value: fahrenheit
      - key: output-unit
        value: celsius
      - key: precision
        value: "2"
      - key: enable-filtering
        value: "true"

Pary klucz-wartość konfiguracji są przekazywane do modułu WASM w czasie wykonywania. Moduł może uzyskać dostęp do tych wartości, aby dostosować jego zachowanie. Takie podejście umożliwia:

• Wdrażanie tego samego modułu WASM z różnymi konfiguracjami
• Dostosowywanie parametrów przetwarzania bez ponownego kompilowania modułów
• Włączanie lub wyłączanie funkcji na podstawie wymagań dotyczących wdrożenia
• Ustawianie wartości specyficznych dla środowiska, takich jak progi lub punkty końcowe

Węzły docelowe

Węzły docelowe definiują miejsce wysyłania przetworzonych danych. Łączą się z punktami końcowymi przepływu danych, które wysyłają dane do brokerów MQTT, magazynu w chmurze lub innych systemów. Każdy węzeł docelowy określa:

• Odwołanie do punktu końcowego wskazującego skonfigurowany punkt końcowy przepływu danych
• Miejsce docelowe danych jako konkretny temat, ścieżka lub lokalizacja dla danych wyjściowych
• Ustawienia schematu wyjściowego (opcjonalnie), które definiują format serializacji i walidację schematu

W przypadku miejsc docelowych magazynu, takich jak Azure Data Lake lub Fabric OneLake, można określić ustawienia schematu wyjściowego, aby kontrolować sposób serializacji i weryfikowania danych.

{
  nodeType: 'Destination'
  name: 'cloud-storage-destination'
  destinationSettings: {
    endpointRef: 'azure-storage-endpoint'
    dataDestination: 'processed-data/temperature'
    outputSchemaSettings: {
      serializationFormat: 'Parquet'
      schemaRef: 'temperature-output-schema:1'
    }
  }
}

- nodeType: Destination
  name: cloud-storage-destination
  destinationSettings:
    endpointRef: azure-storage-endpoint
    dataDestination: processed-data/temperature
    outputSchemaSettings:
      serializationFormat: Parquet
      schemaRef: temperature-output-schema:1

Połączenia węzłów

Połączenia węzłów definiują ścieżkę przepływu danych między węzłami. Każde połączenie określa węzeł źródłowy i węzeł docelowy, tworząc potok przetwarzania. Połączenia mogą opcjonalnie obejmować walidację schematu, aby zapewnić integralność danych między etapami przetwarzania.

Po określeniu weryfikacji schematu system weryfikuje format danych i strukturę, gdy przepływa między węzłami. Walidacja ułatwia wczesne przechwytywanie niespójności danych i zapewnia, że moduły WASM odbierają dane w oczekiwanym formacie.

nodeConnections: [
  {
    from: {
      name: 'sensor-data-source'
      schema: {
        schemaRef: 'sensor-input-schema:1'
        serializationFormat: 'Json'
      }
    }
    to: {
      name: 'temperature-processor'
    }
  }
  {
    from: {
      name: 'temperature-processor'
    }
    to: {
      name: 'cloud-storage-destination'
    }
  }
]

nodeConnections:
  - from:
      name: sensor-data-source
      schema:
        schemaRef: sensor-input-schema:1
        serializationFormat: Json
    to:
      name: temperature-processor
  - from:
      name: temperature-processor
    to:
      name: cloud-storage-destination

Punkty końcowe przepływu danych

Wykresy przepływu danych łączą się z systemami zewnętrznymi za pośrednictwem punktów końcowych przepływu danych. Typ punktu końcowego określa, czy może być używany jako źródło, miejsce docelowe, czy oba:

Punkty końcowe MQTT

Punkty końcowe MQTT mogą służyć zarówno jako źródła, jak i miejsca docelowe. Łączą się z brokerami MQTT, w tym:

• Lokalny broker MQTT operacji usługi Azure IoT (wymagany w każdym przepływie danych)
• Azure Event Grid MQTT
• Niestandardowe brokery MQTT

Aby uzyskać szczegółowe informacje o konfiguracji, zobacz Konfigurowanie punktów końcowych przepływu danych MQTT.

Punkty końcowe platformy Kafka

Punkty końcowe platformy Kafka mogą służyć zarówno jako źródła, jak i miejsca docelowe. Łączą się one z systemami zgodnymi z platformą Kafka, w tym:

• Azure Event Hubs (zgodne z platformą Kafka)
• Klastry platformy Apache Kafka
• Confluent Cloud

Aby uzyskać szczegółowe informacje o konfiguracji, zobacz Konfigurowanie punktów końcowych przepływu danych usługi Azure Event Hubs i platformy Kafka.

Punkty końcowe magazynu

Punkty końcowe magazynu mogą służyć tylko jako miejsca docelowe. Łączą się z systemami magazynowania w chmurze na potrzeby długoterminowego przechowywania danych i analizy:

• Azure Data Lake Storage
• Microsoft Fabric OneLake
• Magazyn lokalny

Punkty końcowe magazynu zwykle wymagają ustawień schematu wyjściowego w celu zdefiniowania formatu serializacji danych.

Punkty końcowe rejestru

Punkty końcowe rejestru zapewniają dostęp do rejestrów kontenerów na potrzeby ściągania modułów WASM i definicji grafu. Nie są one używane bezpośrednio w przepływie danych, ale węzły przetwarzania grafu odwołują się do nich.

Aby uzyskać szczegółowe informacje o konfiguracji, zobacz Konfigurowanie punktów końcowych rejestru.

Treści powiązane

• Konfigurowanie punktów końcowych przepływu danych MQTT
• Konfigurowanie punktów końcowych przepływu danych usługi Azure Event Hubs i platformy Kafka
• Konfigurowanie punktów końcowych przepływu danych usługi Azure Data Lake Storage
• Konfigurowanie punktów końcowych przepływu danych usługi Microsoft Fabric OneLake