在Azure IoT 操作中建立數據流圖

資料流程圖是一種可組合的處理管線，會在資料在來源與目的地之間移動時進行轉換。標準資料流程遵循固定的增強、濾波及映射順序。資料流程圖讓你可以任意順序串接轉換，分支成平行路徑，並依時間窗彙整資料。

本文將一步步說明如何建立資料流程圖。關於資料流程圖及可用轉換的概述，請參閱資料流程圖概覽。

目前資料流程圖僅支援 MQTT、Kafka 及 OpenTelemetry 端點。它們不支援其他端點類型，例如資料湖、Microsoft Fabric OneLake、Azure Data Explorer 和 Local Storage。

轉換使用表達式語言來計算數值、測試條件及參考欄位。表達式以位置指代輸入，而非名稱：列表中的 inputs 第一個輸入為 $1，第二個為 $2，依此類推。例如 cToF 的內建函式可轉換及操作這些值。

完整的運算子、函式、資料型態及元資料欄位清單，請參見 Expressions 參考文獻。

先決條件

Azure IoT 操作的一個實例部署在 Kubernetes 叢集中。如需詳細資訊，請參閱部署 Azure IoT 作業。

安裝在你的開發機器上的 Azure CLI。請查看可用的 Azure CLI 擴充功能，以了解使用azure-iot-ops擴充功能所需的最低版本。使用 az --version 檢查您的版本，並視需要更新 az upgrade。更多資訊請參見安裝 Azure CLI。
Azure CLI 的 Azure IoT 操作延伸模組。使用下列命令將延伸模組新增或更新至最新版本：
```
az extension add --upgrade --name azure-iot-ops
```

Azure IoT 操作版本 1.2 或更後版本。
資料流設定檔。您可以使用預設設定檔。
適用於來源和目的地的資料流端點。預設的 MQTT 代理端點適合入門使用。

本文Azure CLI範例使用環境變數，讓你可以設定一次每個值，然後複製貼上指令 as-is。如果你使用的是快速入門的 Azure IoT 操作 Codespaces 環境，這些變數已經為你設定好，可以跳過這個步驟。否則，在執行指令前，先在 shell 裡設定以下環境變數。

以下腳本設定最常用的環境變數：

環境變數	說明
`SUBSCRIPTION_ID`	包含您的 Azure IoT 操作實例的訂閱 ID。
`RESOURCE_GROUP`	包含你的 Azure IoT 操作實例的資源群組名稱。
`AIO_INSTANCE_NAME`	你的 Azure IoT 操作實例名稱。要列出你的實例，請執行 `az iot ops list -o table`。
`CLUSTER_NAME`	這是支援 Azure Arc 的 Kubernetes 叢集名稱，該叢集負責承載你的實例。
`LOCATION`	例如，使用新資源`eastus`的 Azure 區域。

巴什
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>"

你只需要設定這篇文章所使用的變數。本文可能會使用額外環境變數來表示你選擇的資源名稱。文章說明了如何將它們放置在引入的位置。

本文也使用 GRAPH_NAME 和 PROFILE 環境變數作為資料流圖和輪廓名稱。在執行相關指令前，先設定好每一個。

建立資料流程圖

資料流程圖包含三種元素：將資料帶入的來源、處理資料的轉換，以及將資料送出的 目的地 。依照你想要的資料流動順序連接它們。

在操作體驗中，請前往你的 Azure IoT 操作實例。
選擇 資料流程圖>建立資料流程圖。
輸入資料流圖的名稱並選擇資料流剖面。系統預設會選取預設設定檔。
透過在畫布中加入元素來建立你的管線：
1. 新增來源：選擇來源端點，並設定要訂閱的主題以接收收到的訊息。
2. 新增轉換：選擇一個或多個轉換來處理資料。可用的轉換包括 map、filter、branch、concatenate、window 和 throttle。關於每種轉換類型的詳細資訊，請參閱資料流程圖概述。
3. 新增目的地：選擇目的地端點，並設定傳送處理資料的主題或路徑。
依照你想要的資料流動順序連接元素。
選擇儲存以部署資料流程圖。

Azure CLI 使用一個包含所有節點與連線的單一 JSON 設定檔的資料流程圖。用來 az iot ops dataflowgraph apply 建立或替換圖表。以下範例讀取溫度資料，將其轉換為華氏度，並傳送至目的地主題。

建立 graph.json 一個帶有資料流程圖屬性的檔案。在檔案中 graph.json ，將每個轉換的規則以 excessed JSON 字串的形式儲存在 value 欄位中。關於每個轉換規則的可讀形式，請參閱該轉換類型的操作說明。

{
  "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"
      }
    }
  ]
}

套用設定檔。

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

該圖使用 default 資料流程剖面。若要使用不同的設定檔，請加入 --profile <PROFILE_NAME>。

建立一個包含以下結構的 Bicep .bicep 檔案。此範例建立一個資料流圖，讀取溫度資料，轉換為華氏度，並傳送至目的地主題。

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' }
      }
    ]
  }
}

部署 Bicep 檔案：

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

這很重要

正式生產環境不支援使用 Kubernetes 部署資訊清單，且僅應用於偵錯與測試。

建立一個具有以下結構的 Kubernetes 清單 .yaml 檔案。此範例建立一個資料流圖，讀取溫度資料，轉換為華氏度，並傳送至目的地主題。

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

套用資訊清單：

kubectl apply -f main.yaml

設定來源

資料來源定義了資料進入管線的位置。指定一個端點參考及一個或多個主題。

在資料流程圖編輯器中，選擇來源元素並設定：

Setting	說明
端點	使用的資料流端點。為本機 MQTT 代理程式選取預設。
Topics	訂閱一個或多個主題以接收訊息。

CLI 會一次套用整個圖形，所以請在設定Source檔中將來源設定為graph.json節點，然後執行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'
    ]
  }
}

這很重要

正式生產環境不支援使用 Kubernetes 部署資訊清單，且僅應用於偵錯與測試。

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

新增變換

轉換來源與目的地之間的流程資料。每個轉換都會參考一個內建的產物，並有設定規則。

可用的內建變換有：

轉換	成品	說明
地圖	`azureiotoperations/graph-dataflow-map:1.0.0`	重新命名、重組、計算及複製欄位。
篩選	`azureiotoperations/graph-dataflow-filter:1.0.0`	捨棄符合條件的訊息。
分支	`azureiotoperations/graph-dataflow-branch:1.0.0`	根據條件將每個訊息路由到 `true` 一個或 `false` 路徑。
Concatenate	`azureiotoperations/graph-dataflow-concatenate:1.0.0`	將兩條或以上路徑合併回一條。
窗	`azureiotoperations/graph-dataflow-window:1.0.0`	在一段時間內收集訊息，然後彙整。
節流	`azureiotoperations/graph-dataflow-throttle:1.0.0`	限制每個 MQTT 主題模式的訊息速率。

欲了解更多關於以外部資料豐富訊息的資訊，請參見「以外部資料豐富」。

在資料流程圖編輯器中，選擇 新增轉換 並選擇轉換類型。在視覺編輯器中設定規則。

每個轉換都是一個節點，您可在 nodeType 設定檔中將 Graph 設為 graph.json。轉換的規則是一個 JSON 物件，就像這張將溫度轉換成華氏度的映射：

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

屬性將 configuration 這些規則視為字串，因此跳脫規則 JSON 並將其放在 value 欄位中。使用 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)\"}]}"
      }
    ]
  }
}

Tip

要產生轉義字串，將規則儲存到像 rules.json、執行 jq -c . rules.json這樣的檔案，然後將單行輸出貼到欄位中 value 。

每個變換都是具有 nodeType: 'Graph' 的節點。這個屬性以 configuration 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)"}]}'
      }
    ]
  }
}

這很重要

正式生產環境不支援使用 Kubernetes 部署資訊清單，且僅應用於偵錯與測試。

每個變換都是具有 nodeType: Graph 的節點。這個屬性以 configuration 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)"
              }
            ]
          }

串連多個轉換

串接任意數量的轉換。在 nodeConnections 區段中，以您希望資料流動的順序連接它們。

在畫布上拖曳轉換之間的連結來定義處理順序。

請在你的 nodeConnections 設定檔的 graph.json 區段中定義處理順序：

"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' } }
]

這很重要

正式生產環境不支援使用 Kubernetes 部署資訊清單，且僅應用於偵錯與測試。

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

設定目的地

目的地定義了處理後的資料去向。指定端點參考及主題或路徑。

選擇目的地元素並設定：

Setting	說明
端點	用於傳送資料的資料流端點。
Topic	發佈處理後資料的主題或路徑。

在您的 Destination 設定檔中將目的地設為 graph.json 節點，然後使用 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'
  }
}

這很重要

正式生產環境不支援使用 Kubernetes 部署資訊清單，且僅應用於偵錯與測試。

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

關於基於訊息內容的動態主題路由，請參見「將訊息導向不同主題」。

確認資料流程圖是否正常運作

部署資料流程圖後，請確認它是否在執行：

在操作體驗中，選擇你的資料流圖以查看其狀態。健康的圖表顯示出運行狀態。

請使用 az iot ops dataflowgraph show 以查看圖表細節：

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

要列出所有與某一輪廓相關的資料流圖，請使用 az iot ops dataflowgraph list：

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

請檢查資源 DataflowGraph 狀態：

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

這很重要

正式生產環境不支援使用 Kubernetes 部署資訊清單，且僅應用於偵錯與測試。

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

請查看Pod日誌是否有錯誤：

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

意見反應

此頁面對您有幫助嗎？

上次更新時間： 2026-08-14

在Azure IoT 操作中建立數據流圖

先決條件

建立資料流程圖

設定來源

新增變換

串連多個轉換

設定目的地

確認資料流程圖是否正常運作

相關內容

意見反應

其他資源