Tworzenie grafu przepływu danych w operacjach usługi Azure IoT

Wykres przepływu danych to potok przetwarzania, który przekształca dane w miarę przemieszczania się między źródłami i miejscami docelowymi. Standardowy przepływ danych przebiega według stałej sekwencji: wzbogacenie, filtracja, mapowanie. Wykres przepływu danych umożliwia tworzenie łańcuchów przekształceń w dowolnej kolejności, rozgałęzianie w ścieżkach równoległych i agregowanie danych w oknach czasu.

W tym artykule opisano krok po kroku tworzenie grafu przepływu danych. Aby zapoznać się z omówieniem wykresów przepływu danych i dostępnych przekształceń, zobacz Omówienie wykresów przepływu danych.

Wykresy przepływu danych obsługują obecnie tylko punkty końcowe MQTT, Kafka i OpenTelemetry. Nie obsługują innych typów endpointów, takich jak Data Lake, Microsoft Fabric, OneLake, Azure Data Explorer czy lokalna pamięć masowa.

Transformacje wykorzystują język wyrażeń do obliczania wartości, warunków testowych i pól odniesienia. Wyrażenia odnoszą się do wejść według położenia, a nie nazwy: pierwsze wejście w inputs liście to $1, drugie to $2, i tak dalej. Wbudowane funkcje, takie jak cToF, konwertują i modyfikują te wartości.

Pełną listę operatorów, funkcji, typów danych i pól metadanych można znaleźć w sekcji Expressions.

Wymagania wstępne

Instancja usługi Operacje Azure IoT wdrożona w klastrze Kubernetes. Aby uzyskać więcej informacji, zobacz Deploy Operacje Azure IoT.

Na komputerze deweloperskim zainstalowano Azure CLI. Sprawdź dostępne rozszerzenia Azure CLI dla minimalnej wymaganej wersji do korzystania z rozszerzenia azure-iot-ops. Użyj az --version do sprawdzenia swojej wersji i az upgrade do aktualizacji, jeśli to konieczne. Aby uzyskać więcej informacji, zobacz Install the Azure CLI .
Rozszerzenie Operacje Azure IoT dla Azure CLI. Użyj następującego polecenia, aby dodać rozszerzenie lub zaktualizować je do najnowszej wersji:
```
az extension add --upgrade --name azure-iot-ops
```

Operacje Azure IoT w wersji 1.2 lub nowszej.
Profil przepływu danych. Możesz użyć profilu domyślnego.
Punkt końcowy przepływu danych dla źródła i miejsca docelowego. Domyślny punkt końcowy brokera MQTT działa dobrze na początek.

Azure CLI przykładów w tym artykule używa zmiennych środowiskowych, dzięki czemu można ustawić każdą wartość raz, a następnie skopiować i wkleić polecenia as-is. Jeśli korzystasz ze środowiska Operacje Azure IoT Codespaces z quickstartu, te zmienne są już ustawione i możesz pominąć ten krok. W przeciwnym razie ustaw następujące zmienne środowiskowe w swojej poskoczce przed uruchomieniem poleceń.

Poniższe skrypty określają najczęściej używane zmienne środowiskowe:

Zmienna środowiskowa	Opis
`SUBSCRIPTION_ID`	ID subskrypcji zawierającej Twoją instancję Operacje Azure IoT.
`RESOURCE_GROUP`	Nazwa grupy zasobów zawierającej Twoją instancję Operacje Azure IoT.
`AIO_INSTANCE_NAME`	Nazwa Twojej instancji Operacje Azure IoT. Aby wypisać swoje instancje, uruchom `az iot ops list -o table`.
`CLUSTER_NAME`	Nazwa klastra Kubernetes z włączonym Azure Arc, który hostuje Twoją instancję.
`LOCATION`	Region Azure do wykorzystania dla nowych zasobów, na przykład `eastus`.

Bash
PowerShell

SUBSCRIPTION_ID=<subscription-id>
RESOURCE_GROUP=<resource-group-name>
AIO_INSTANCE_NAME=<instance-name>
CLUSTER_NAME=<cluster-name>
LOCATION=<region>

$SUBSCRIPTION_ID = "<subscription-id>"
$RESOURCE_GROUP = "<resource-group-name>"
$AIO_INSTANCE_NAME = "<instance-name>"
$CLUSTER_NAME = "<cluster-name>"
$LOCATION = "<region>"

Wystarczy ustawić zmienne, których używa ten artykuł. W tym artykule może być użytych dodatkowych zmiennych środowiskowych do wyboru nazw zasobów. Artykuł wyjaśnia, jak ustawić je tam, gdzie są wprowadzane.

W tym artykule używa również zmiennych środowiskowych i GRAPH_NAMEPROFILE środowiskowych dla nazw wykresu przepływu danych oraz profili. Ustaw każdą z nich przed uruchomieniem odpowiednich poleceń.

Tworzenie grafu przepływu danych

Wykres przepływu danych zawiera trzy typy elementów: źródła , które przesyłają dane, przekształcają je, a lokalizacje docelowe , które je wysyłają. Połącz je w kolejności, w której dane mają być przepływane.

W doświadczeniu operacyjnym przejdź do instancji Operacje Azure IoT.
Wybierz Wykres przepływu danych>Utwórz wykres przepływu danych.
Wprowadź nazwę grafu przepływu danych i wybierz profil przepływu danych. Domyślny profil jest domyślnie wybierany.
Zbuduj swój pipeline, dodając elementy do kanwy.
1. Dodaj źródło: wybierz źródłowy punkt końcowy i skonfiguruj tematy, aby subskrybować komunikaty przychodzące.
2. Dodawanie przekształceń: wybierz co najmniej jedną transformację, aby przetworzyć dane. Dostępne transformacje to mapa, filtr, rozgałęzienie, konkatenacja, okno oraz przepustnica. Aby uzyskać szczegółowe informacje na temat każdego typu transformacji, zobacz Omówienie wykresów przepływu danych.
3. Dodaj miejsce docelowe: wybierz docelowy punkt końcowy i skonfiguruj temat lub ścieżkę do wysyłania przetworzonych danych.
Połącz elementy w kolejności, w której dane mają być przepływane.
Wybierz pozycję Zapisz , aby wdrożyć wykres przepływu danych.

Azure CLI wykorzystuje graf przepływu danych z pojedynczego pliku konfiguracyjnego JSON, który zawiera wszystkie węzły i połączenia. Użyj az iot ops dataflowgraph apply polecenia , aby utworzyć lub zamienić wykres. Poniższy przykład odczytuje dane temperaturowe, konwertuje je na Fahrenheita i wysyła do tematu docelowego.

graph.json Utwórz plik z właściwościami grafu przepływu danych. W graph.json pliku przechowuj reguły każdej transformacji value w polu jako ucieczkę łańcucha JSON. Aby zapoznać się z czytelną formą reguł każdego przekształcenia, zobacz instrukcje dotyczące tego typu transformacji.

{
  "mode": "Enabled",
  "nodes": [
    {
      "nodeType": "Source",
      "name": "sensors",
      "sourceSettings": {
        "endpointRef": "default",
        "dataSources": [
          "telemetry/temperature"
        ]
      }
    },
    {
      "nodeType": "Graph",
      "name": "convert",
      "graphSettings": {
        "registryEndpointRef": "default",
        "artifact": "azureiotoperations/graph-dataflow-map:1.0.0",
        "configuration": [
          {
            "key": "rules",
            "value": "{\"map\":[{\"inputs\":[\"*\"],\"output\":\"*\"},{\"inputs\":[\"temperature\"],\"output\":\"temperature_f\",\"expression\":\"cToF($1)\"}]}"
          }
        ]
      }
    },
    {
      "nodeType": "Destination",
      "name": "output",
      "destinationSettings": {
        "endpointRef": "default",
        "dataDestination": "telemetry/converted"
      }
    }
  ],
  "nodeConnections": [
    {
      "from": {
        "name": "sensors"
      },
      "to": {
        "name": "convert"
      }
    },
    {
      "from": {
        "name": "convert"
      },
      "to": {
        "name": "output"
      }
    }
  ]
}

Zastosuj plik konfiguracji.

az iot ops dataflowgraph apply \
  --name temperature-processing \
  --instance $AIO_INSTANCE_NAME \
  --resource-group $RESOURCE_GROUP \
  --config-file graph.json

Wykres wykorzystuje default profil przepływu danych. Aby użyć innego profilu, dodaj polecenie --profile <PROFILE_NAME>.

Utwórz plik Bicep .bicep z następującą strukturą. W tym przykładzie tworzony jest graf przepływu danych, który odczytuje dane temperatury, konwertuje je na fahrenheita i wysyła go do tematu docelowego.

param aioInstanceName string = '<AIO_INSTANCE_NAME>'
param customLocationName string = '<CUSTOM_LOCATION_NAME>'

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

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

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

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2026-07-01' = {
  parent: defaultDataflowProfile
  name: 'temperature-processing'
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    profileRef: 'default'
    mode: 'Enabled'
    nodes: [
      {
        nodeType: 'Source'
        name: 'sensors'
        sourceSettings: {
          endpointRef: 'default'
          dataSources: [
            'telemetry/temperature'
          ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'convert'
        graphSettings: {
          registryEndpointRef: 'default'
          artifact: 'azureiotoperations/graph-dataflow-map:1.0.0'
          configuration: [
            {
              key: 'rules'
              value: '{"map":[{"inputs":["*"],"output":"*"},{"inputs":["temperature"],"output":"temperature_f","expression":"cToF($1)"}]}'
            }
          ]
        }
      }
      {
        nodeType: 'Destination'
        name: 'output'
        destinationSettings: {
          endpointRef: 'default'
          dataDestination: 'telemetry/converted'
        }
      }
    ]
    nodeConnections: [
      {
        from: { name: 'sensors' }
        to: { name: 'convert' }
      }
      {
        from: { name: 'convert' }
        to: { name: 'output' }
      }
    ]
  }
}

Wdróż plik Bicep:

az deployment group create --resource-group $RESOURCE_GROUP --template-file main.bicep

Ważna

Korzystanie z manifestów wdrożenia platformy Kubernetes nie jest obsługiwane w środowiskach produkcyjnych i powinno być używane tylko do debugowania i testowania.

Utwórz plik manifestu .yaml kubernetes z następującą strukturą. W tym przykładzie tworzony jest graf przepływu danych, który odczytuje dane temperatury, konwertuje je na fahrenheita i wysyła go do tematu docelowego.

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
  name: temperature-processing
  namespace: azure-iot-operations
spec:
  profileRef: default
  nodes:
    - nodeType: Source
      name: sensors
      sourceSettings:
        endpointRef: default
        dataSources:
          - telemetry/temperature

    - nodeType: Graph
      name: convert
      graphSettings:
        registryEndpointRef: default
        artifact: azureiotoperations/graph-dataflow-map:1.0.0
        configuration:
          - key: rules
            value: |
              {
                "map": [
                  {
                    "inputs": ["*"],
                    "output": "*"
                  },
                  {
                    "inputs": ["temperature"],
                    "output": "temperature_f",
                    "expression": "cToF($1)"
                  }
                ]
              }

    - nodeType: Destination
      name: output
      destinationSettings:
        endpointRef: default
        dataDestination: telemetry/converted

  nodeConnections:
    - from:
        name: sensors
      to:
        name: convert
    - from:
        name: convert
      to:
        name: output

Zastosuj manifest:

kubectl apply -f main.yaml

Konfigurowanie źródła

Źródło definiuje miejsce, w którym dane wchodzą do potoku. Określ odwołanie do punktu końcowego i co najmniej jeden temat.

W edytorze grafu przepływu danych wybierz element źródłowy i skonfiguruj:

Setting	Opis
Punkt końcowy	Punkt końcowy przepływu danych do użycia. Wybierz wartość domyślną dla lokalnego brokera MQTT.
Topics	Co najmniej jeden subskrybowany temat do wiadomości przychodzących.

CLI przetwarza cały graf naraz, więc skonfiguruj źródło jako węzeł Source w pliku konfiguracji graph.json, a następnie uruchom az iot ops dataflowgraph apply:

{
  "nodeType": "Source",
  "name": "sensors",
  "sourceSettings": {
    "endpointRef": "default",
    "dataSources": [
      "telemetry/temperature",
      "telemetry/humidity"
    ]
  }
}

{
  nodeType: 'Source'
  name: 'sensors'
  sourceSettings: {
    endpointRef: 'default'
    dataSources: [
      'telemetry/temperature'
      'telemetry/humidity'
    ]
  }
}

Ważna

Korzystanie z manifestów wdrożenia platformy Kubernetes nie jest obsługiwane w środowiskach produkcyjnych i powinno być używane tylko do debugowania i testowania.

- nodeType: Source
  name: sensors
  sourceSettings:
    endpointRef: default
    dataSources:
      - telemetry/temperature
      - telemetry/humidity

Dodawanie przekształceń

Przekształca dane procesowe między źródłem a celem. Każda transformacja odnosi się do wbudowanego artefaktu i posiada reguły konfiguracyjne.

Dostępne wbudowane przekształcenia to:

Przekształć	Artifact	Opis
Mapa	`azureiotoperations/graph-dataflow-map:1.0.0`	Zmienianie nazwy, restrukturyzacja, obliczenia i kopiowanie pól.
Filtr	`azureiotoperations/graph-dataflow-filter:1.0.0`	Usuwanie komunikatów pasujących do warunku.
Gałąź	`azureiotoperations/graph-dataflow-branch:1.0.0`	Kieruj każdy komunikat do ścieżki `true` lub `false` na podstawie warunku.
Złączyć	`azureiotoperations/graph-dataflow-concatenate:1.0.0`	Scal dwie lub więcej ścieżek ponownie w jedną.
Okno	`azureiotoperations/graph-dataflow-window:1.0.0`	Zbierz komunikaty w przedziale czasu, a następnie zagreguj.
Ograniczanie	`azureiotoperations/graph-dataflow-throttle:1.0.0`	Ogranicz częstotliwość wiadomości dla wzorca tematu MQTT.

Aby uzyskać więcej informacji na temat wzbogacania komunikatów o dane zewnętrzne, zobacz Wzbogacanie o dane zewnętrzne.

W edytorze wykresu przepływu danych wybierz pozycję Dodaj przekształcenie i wybierz typ przekształcenia. Skonfiguruj reguły w edytorze wizualizacji.

Każda transformacja to węzeł z ustawieniem nodeType na Graph w pliku konfiguracyjnym graph.json. Reguły przekształcenia są obiektem JSON, takim jak ta mapa, która konwertuje temperaturę na Fahrenheit:

{
  "map": [
    {
      "inputs": ["temperature"],
      "output": "temperature_f",
      "expression": "cToF($1)"
    }
  ]
}

Własność przyjmuje configuration te reguły jako ciąg rzeczy, więc uciekaj value z reguł JSON i umieść ją w polu. Zastosuj pełny graf za pomocą polecenia az iot ops dataflowgraph apply:

{
  "nodeType": "Graph",
  "name": "convert",
  "graphSettings": {
    "registryEndpointRef": "default",
    "artifact": "azureiotoperations/graph-dataflow-map:1.0.0",
    "configuration": [
      {
        "key": "rules",
        "value": "{\"map\":[{\"inputs\":[\"temperature\"],\"output\":\"temperature_f\",\"expression\":\"cToF($1)\"}]}"
      }
    ]
  }
}

Wskazówka

Aby wygenerować ciąg escapowany, zapisz reguły w pliku takim jak rules.json, uruchom jq -c . rules.json, i wklej jednolinijowy wynik do value pola.

Każda transformacja jest węzłem wykorzystującym nodeType: 'Graph'. Właściwość configuration przekazuje reguły jako ciąg JSON:

{
  nodeType: 'Graph'
  name: 'convert'
  graphSettings: {
    registryEndpointRef: 'default'
    artifact: 'azureiotoperations/graph-dataflow-map:1.0.0'
    configuration: [
      {
        key: 'rules'
        value: '{"map":[{"inputs":["temperature"],"output":"temperature_f","expression":"cToF($1)"}]}'
      }
    ]
  }
}

Ważna

Korzystanie z manifestów wdrożenia platformy Kubernetes nie jest obsługiwane w środowiskach produkcyjnych i powinno być używane tylko do debugowania i testowania.

Każda transformacja jest węzłem wykorzystującym nodeType: Graph. Właściwość configuration przekazuje reguły jako ciąg JSON:

- nodeType: Graph
  name: convert
  graphSettings:
    registryEndpointRef: default
    artifact: azureiotoperations/graph-dataflow-map:1.0.0
    configuration:
      - key: rules
        value: |
          {
            "map": [
              {
                "inputs": ["temperature"],
                "output": "temperature_f",
                "expression": "cToF($1)"
              }
            ]
          }

Łączenie wielu przekształceń

Łańcuch dowolnej liczby transformacji. Połącz je w sekcji nodeConnections w kolejności, w jakiej ma przepływać dane.

Przeciągnij połączenia między przekształceniami na kanwie, aby zdefiniować kolejność przetwarzania.

Zdefiniuj kolejność przetwarzania w nodeConnections sekcji graph.json pliku konfiguracji:

"nodeConnections": [
  {
    "from": {
      "name": "sensors"
    },
    "to": {
      "name": "remove-bad-data"
    }
  },
  {
    "from": {
      "name": "remove-bad-data"
    },
    "to": {
      "name": "convert"
    }
  },
  {
    "from": {
      "name": "convert"
    },
    "to": {
      "name": "output"
    }
  }
]

nodeConnections: [
  { from: { name: 'sensors' }, to: { name: 'remove-bad-data' } }
  { from: { name: 'remove-bad-data' }, to: { name: 'convert' } }
  { from: { name: 'convert' }, to: { name: 'output' } }
]

Ważna

Korzystanie z manifestów wdrożenia platformy Kubernetes nie jest obsługiwane w środowiskach produkcyjnych i powinno być używane tylko do debugowania i testowania.

nodeConnections:
  - from: { name: sensors }
    to: { name: remove-bad-data }
  - from: { name: remove-bad-data }
    to: { name: convert }
  - from: { name: convert }
    to: { name: output }

Konfigurowanie miejsca docelowego

Miejsce docelowe definiuje, gdzie trafiają przetwarzane dane. Określ odwołanie do punktu końcowego i temat lub ścieżkę.

Wybierz element docelowy i skonfiguruj:

Setting	Opis
Punkt końcowy	Punkt końcowy przepływu danych do wysyłania danych.
Topic	Temat lub ścieżka do publikowania przetworzonych danych.

Skonfiguruj miejsce docelowe jako węzeł Destination w pliku konfiguracji graph.json, a następnie zastosuj cały graf za pomocą polecenia az iot ops dataflowgraph apply:

{
  "nodeType": "Destination",
  "name": "output",
  "destinationSettings": {
    "endpointRef": "default",
    "dataDestination": "telemetry/processed"
  }
}

{
  nodeType: 'Destination'
  name: 'output'
  destinationSettings: {
    endpointRef: 'default'
    dataDestination: 'telemetry/processed'
  }
}

Ważna

Korzystanie z manifestów wdrożenia platformy Kubernetes nie jest obsługiwane w środowiskach produkcyjnych i powinno być używane tylko do debugowania i testowania.

- nodeType: Destination
  name: output
  destinationSettings:
    endpointRef: default
    dataDestination: telemetry/processed

Aby uzyskać dynamiczny routing tematów oparty na zawartości komunikatów, zobacz Kierowanie komunikatów do różnych tematów.

Sprawdź, czy wykres przepływu danych działa

Po wdrożeniu grafu przepływu danych sprawdź, czy działa:

W doświadczeniu operacyjnym wybierz wykres przepływu danych, aby zobaczyć jego status. Wykres w dobrej kondycji oznacza stan Uruchomiony.

Użyj az iot ops dataflowgraph show polecenia , aby wyświetlić szczegóły grafu:

az iot ops dataflowgraph show \
  --name temperature-processing \
  --instance $AIO_INSTANCE_NAME \
  --resource-group $RESOURCE_GROUP

Aby wyświetlić listę wszystkich wykresów przepływu danych skojarzonych z profilem, użyj polecenia az iot ops dataflowgraph list:

az iot ops dataflowgraph list \
  --instance $AIO_INSTANCE_NAME \
  --resource-group $RESOURCE_GROUP

Sprawdź stan DataflowGraph zasobu:

az resource show --resource-group $RESOURCE_GROUP --resource-type Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs --name $GRAPH_NAME --parent instances/$AIO_INSTANCE_NAME/dataflowProfiles/$PROFILE

Ważna

Korzystanie z manifestów wdrożenia platformy Kubernetes nie jest obsługiwane w środowiskach produkcyjnych i powinno być używane tylko do debugowania i testowania.

kubectl get dataflowgraph temperature-processing -n azure-iot-operations

Sprawdź dzienniki pod pod kątem błędów:

kubectl logs -l app=dataflow -n azure-iot-operations --tail=50

Opinia

Czy ta strona była pomocna?

Last updated on 2026-07-31

Tworzenie grafu przepływu danych w operacjach usługi Azure IoT

Wymagania wstępne

Tworzenie grafu przepływu danych

Konfigurowanie źródła

Dodawanie przekształceń

Łączenie wielu przekształceń

Konfigurowanie miejsca docelowego

Sprawdź, czy wykres przepływu danych działa

Treści powiązane

Opinia

Dodatkowe zasoby