Anreichern von Daten mit externen Datasets in Datenflussdiagrammen

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

Fügen Sie in der Transformationskonfiguration ein Dataset hinzu. Konfigurieren:

Die CLI wendet das gesamte Diagramm aus einer Konfigurationsdatei an. Füge diese Regeln den Transformknoten configuration in Your graph.json hinzu und wende sie mit .az iot ops dataflowgraph apply

Die Regeln sind ein JSON-Objekt:

{
  "datasets": [
    {
      "key": "device-metadata as device",
      "inputs": ["$source.deviceId", "$context.deviceId"],
      "expression": "$1 == $2"
    }
  ],
  "map": [
    {
      "inputs": ["$context(device).displayName"],
      "output": "deviceName"
    }
  ]
}

Diese Regeln werden als Escapezeichenfolge im value-Feld abgelegt:

"configuration": [
  {
    "key": "rules",
    "value": "{\"datasets\":[{\"key\":\"device-metadata as device\",\"inputs\":[\"$source.deviceId\",\"$context.deviceId\"],\"expression\":\"$1 == $2\"}],\"map\":[{\"inputs\":[\"$context(device).displayName\"],\"output\":\"deviceName\"}]}"
  }
]

Die Datasetkonfiguration ist Teil der Regeln JSON:

configuration: [
  {
    key: 'rules'
    value: '{"datasets":[{"key":"device-metadata as device","inputs":["$source.deviceId","$context.deviceId"],"expression":"$1 == $2"}],"map":[{"inputs":["$context(device).displayName"],"output":"deviceName"}]}'
  }
]

{
  "datasets": [
    {
      "key": "device-metadata as device",
      "inputs": ["$source.deviceId", "$context.deviceId"],
      "expression": "$1 == $2"
    }
  ],
  "map": [
    {
      "inputs": ["$context(device).displayName"],
      "output": "deviceName"
    }
  ]
}

In der Transformationskonfiguration fügen Sie einen Datensatz hinzu und konfigurieren:

Fügen Sie im Datensatzeintrag dynamicValues neben key hinzu:

{
  "datasets": [
    {
      "key": "calibration:$1 as calibration",
      "dynamicValues": ["sensorId"],
      "inputs": ["$source.sensorId", "$context.sensorId"],
      "expression": "$1 == $2"
    }
  ],
  "map": [
    {
      "inputs": ["$context(calibration).offset"],
      "output": "calibrationOffset"
    }
  ]
}

Fügen Sie diese Eigenschaft zur configuration des Transformationsknotens in Ihrem graph.json als Zeichenfolge mit Escapezeichen im Feld value hinzu. Wenden Sie es dann mit az iot ops dataflowgraph apply an.

"configuration": [
  {
    "key": "rules",
    "value": "{\"datasets\":[{\"key\":\"calibration:$1 as calibration\",\"dynamicValues\":[\"sensorId\"],\"inputs\":[\"$source.sensorId\",\"$context.sensorId\"],\"expression\":\"$1 == $2\"}],\"map\":[{\"inputs\":[\"$context(calibration).offset\"],\"output\":\"calibrationOffset\"}]}"
  }
]

configuration: [
  {
    key: 'rules'
    value: '{"datasets":[{"key":"calibration:$1 as calibration","dynamicValues":["sensorId"],"inputs":["$source.sensorId","$context.sensorId"],"expression":"$1 == $2"}],"map":[{"inputs":["$context(calibration).offset"],"output":"calibrationOffset"}]}'
  }
]

datasets:
  - key: "calibration:$1 as calibration"
    dynamicValues:
      - sensorId
    inputs:
      - $source.sensorId
      - $context.sensorId
    expression: "$1 == $2"

map:
  - inputs:
      - "$context(calibration).offset"
    output: calibrationOffset

Fügen Sie Zuordnungsregeln hinzu, die auf erweiterte Felder verweisen:

Die angereicherten Feldreferenzen sind Teil der Zuordnungsregeln in Ihrer graph.json-Konfigurationsdatei. Füge die Regeln dem Transformationsknoten configuration hinzu und wende den Graphen mit az iot ops dataflowgraph apply an:

"map": [
  {
    "inputs": [
      "$context(position).WorkingHours"
    ],
    "output": "WorkingHours"
  },
  {
    "inputs": [
      "rawValue",
      "$context(product).multiplier"
    ],
    "output": "adjustedValue",
    "expression": "$1 * $2"
  }
]

Die erweiterten Feldverweise sind Teil der JSON-Zuordnungsregeln:

'{"datasets":[...],"map":[{"inputs":["$context(position).WorkingHours"],"output":"WorkingHours"},{"inputs":["rawValue","$context(product).multiplier"],"output":"adjustedValue","expression":"$1 * $2"}]}'

- inputs:
    - "$context(position).WorkingHours"
  output: WorkingHours

- inputs:
    - rawValue                          # $1
    - "$context(product).multiplier"    # $2
  output: adjustedValue
  expression: "$1 * $2"

Fügen Sie eine Filterregel mit den Eingaben rawValue, $context(limits).multiplier und $context(limits).baseLimit sowie dem Ausdruck $1 * $2 > $3 hinzu.

Die CLI wendet das gesamte Diagramm aus einer Konfigurationsdatei an. Die folgende Filterregel bezieht sich auf die angereicherten $context(limits) Felder in ihrem Match-Ausdruck, bereitgestellt als JSON-Objekt:

{
  "datasets": [
    {
      "key": "device_limits as limits",
      "inputs": [
        "$source.deviceId",
        "$context.deviceId"
      ],
      "expression": "$1 == $2"
    }
  ],
  "filter": [
    {
      "inputs": [
        "rawValue",
        "$context(limits).multiplier",
        "$context(limits).baseLimit"
      ],
      "expression": "$1 * $2 > $3"
    }
  ]
}

Fügen Sie diese Regeln dem Transformationsknoten configuration in Ihrem graph.json als maskierte Zeichenfolge im Feld value hinzu und wenden Sie sie dann mit az iot ops dataflowgraph apply an:

"configuration": [
  {
    "key": "rules",
    "value": "{\"datasets\":[{\"key\":\"device_limits as limits\",\"inputs\":[\"$source.deviceId\",\"$context.deviceId\"],\"expression\":\"$1 == $2\"}],\"filter\":[{\"inputs\":[\"rawValue\",\"$context(limits).multiplier\",\"$context(limits).baseLimit\"],\"expression\":\"$1 * $2 > $3\"}]}"
  }
]

Die folgende Filterregel bezieht sich auf die angereicherten $context(limits) Felder in den Regeln JSON:

'{"datasets":[{"key":"device_limits as limits","inputs":["$source.deviceId","$context.deviceId"],"expression":"$1 == $2"}],"filter":[{"inputs":["rawValue","$context(limits).multiplier","$context(limits).baseLimit"],"expression":"$1 * $2 > $3"}]}'

{
  "datasets": [
    {
      "key": "device_limits as limits",
      "inputs": ["$source.deviceId", "$context.deviceId"],
      "expression": "$1 == $2"
    }
  ],
  "filter": [
    {
      "inputs": ["rawValue", "$context(limits).multiplier", "$context(limits).baseLimit"],
      "expression": "$1 * $2 > $3"
    }
  ]
}

Konfigurieren Sie eine Verzweigungsregel mit den Eingaben quantity, $context(mult).factor und $context(mult).threshold sowie dem Ausdruck $1 * $2 > $3.

Die CLI wendet das gesamte Diagramm aus einer Konfigurationsdatei an. Die folgende Verzweigungsregel leitet Nachrichten mithilfe der angereicherten $context(mult) Felder weiter, die als JSON-Objekt bereitgestellt werden:

{
  "datasets": [
    {
      "key": "multipliers as mult",
      "inputs": [
        "$source.productCode",
        "$context.productCode"
      ],
      "expression": "$1 == $2"
    }
  ],
  "branch": {
    "inputs": [
      "quantity",
      "$context(mult).factor",
      "$context(mult).threshold"
    ],
    "expression": "$1 * $2 > $3"
  }
}

Fügen Sie diese Regeln dem Transformationsknoten configuration in Ihrem graph.json als maskierte Zeichenfolge im Feld value hinzu und wenden Sie sie dann mit az iot ops dataflowgraph apply an:

"configuration": [
  {
    "key": "rules",
    "value": "{\"datasets\":[{\"key\":\"multipliers as mult\",\"inputs\":[\"$source.productCode\",\"$context.productCode\"],\"expression\":\"$1 == $2\"}],\"branch\":{\"inputs\":[\"quantity\",\"$context(mult).factor\",\"$context(mult).threshold\"],\"expression\":\"$1 * $2 > $3\"}}"
  }
]

Die folgende Zweigregel bezieht sich auf die angereicherten $context(mult) Felder in den Regeln JSON:

'{"datasets":[{"key":"multipliers as mult","inputs":["$source.productCode","$context.productCode"],"expression":"$1 == $2"}],"branch":{"inputs":["quantity","$context(mult).factor","$context(mult).threshold"],"expression":"$1 * $2 > $3"}}'

{
  "datasets": [
    {
      "key": "multipliers as mult",
      "inputs": ["$source.productCode", "$context.productCode"],
      "expression": "$1 == $2"
    }
  ],
  "branch": {
    "inputs": ["quantity", "$context(mult).factor", "$context(mult).threshold"],
    "expression": "$1 * $2 > $3"
  }
}

Fügen Sie eine Kartenregel mit Eingabe $context(device).* und Ausgabe *hinzu.

Die CLI wendet den gesamten Graphen aus einer Konfigurationsdatei an, also füge diese Regel an die entsprechende Stelle in Your graph.json hinzu und wende sie mit az iot ops dataflowgraph applyFolgendes an:

{
  "inputs": [
    "$context(device).*"
  ],
  "output": "*"
}

{
  inputs: [ '$context(device).*' ]
  output: '*'
}

- inputs:
    - "$context(device).*"
  output: "*"

Erstellen Sie in der Betriebsumgebung ein Datenflussdiagramm mit Anreicherung:

1. Fügen Sie eine Quelle hinzu, die aus Ihrem MQTT-Topic gelesen wird.
2. Fügen Sie eine Kartentransformation hinzu. Fügen Sie in der Datasetkonfiguration ein Dataset mit dem Zustandsspeicherschlüssel und der Übereinstimmungsbedingung hinzu.
3. In den Kartenregeln werden angereicherte Felder mithilfe von $context(<alias>).<field> Syntax referenziert.
4. Fügen Sie ein Ziel hinzu, das an Ihr Ausgabe-Topic sendet.

Die Azure CLI verwendet einen Datenflussgraphen aus einer einzelnen JSON-Konfigurationsdatei. Erstellen Sie eine graph.json Datei mit den Diagrammeigenschaften. In der graph.json Datei speichert jede Transformation ihre Regeln im Feld value als entkommene JSON-Zeichenkette. Für die lesbare Form der Regeln jeder Transformation siehe den Anleitungsartikel zu diesem Transformationstyp.

{
  "mode": "Enabled",
  "nodes": [
    {
      "nodeType": "Source",
      "name": "sensors",
      "sourceSettings": {
        "endpointRef": "default",
        "dataSources": [
          "telemetry/sensors"
        ]
      }
    },
    {
      "nodeType": "Graph",
      "name": "enrich-and-map",
      "graphSettings": {
        "registryEndpointRef": "default",
        "artifact": "azureiotoperations/graph-dataflow-map:1.1.0",
        "configuration": [
          {
            "key": "rules",
            "value": "{\"datasets\":[{\"key\":\"device-metadata as device\",\"inputs\":[\"$source.deviceId\",\"$context.deviceId\"],\"expression\":\"$1 == $2\"}],\"map\":[{\"inputs\":[\"*\"],\"output\":\"*\"},{\"inputs\":[\"$context(device).displayName\"],\"output\":\"deviceName\"},{\"inputs\":[\"$context(device).location\"],\"output\":\"location\"}]}"
          }
        ]
      }
    },
    {
      "nodeType": "Destination",
      "name": "output",
      "destinationSettings": {
        "endpointRef": "default",
        "dataDestination": "telemetry/enriched"
      }
    }
  ],
  "nodeConnections": [
    {
      "from": {
        "name": "sensors"
      },
      "to": {
        "name": "enrich-and-map"
      }
    },
    {
      "from": {
        "name": "enrich-and-map"
      },
      "to": {
        "name": "output"
      }
    }
  ]
}

Wenden Sie die Konfigurationsdatei an.

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

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2026-03-01' = {
  name: 'enrich-example'
  parent: dataflowProfile
  properties: {
    mode: 'Enabled'
    nodes: [
      {
        nodeType: 'Source'
        name: 'sensors'
        sourceSettings: {
          endpointRef: 'default'
          dataSources: [ 'telemetry/sensors' ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'enrich-and-map'
        graphSettings: {
          registryEndpointRef: 'default'
          artifact: 'azureiotoperations/graph-dataflow-map:1.1.0'
          configuration: [
            {
              key: 'rules'
              value: '{"datasets":[{"key":"device-metadata as device","inputs":["$source.deviceId","$context.deviceId"],"expression":"$1 == $2"}],"map":[{"inputs":["*"],"output":"*"},{"inputs":["$context(device).displayName"],"output":"deviceName"},{"inputs":["$context(device).location"],"output":"location"}]}'
            }
          ]
        }
      }
      {
        nodeType: 'Destination'
        name: 'output'
        destinationSettings: {
          endpointRef: 'default'
          dataDestination: 'telemetry/enriched'
        }
      }
    ]
    nodeConnections: [
      { from: { name: 'sensors' }, to: { name: 'enrich-and-map' } }
      { from: { name: 'enrich-and-map' }, to: { name: 'output' } }
    ]
  }
}

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

    - nodeType: Graph
      name: enrich-and-map
      graphSettings:
        registryEndpointRef: default
        artifact: azureiotoperations/graph-dataflow-map:1.1.0
        configuration:
          - key: rules
            value: |
              {
                "datasets": [
                  {
                    "key": "device-metadata as device",
                    "inputs": ["$source.deviceId", "$context.deviceId"],
                    "expression": "$1 == $2"
                  }
                ],
                "map": [
                  { "inputs": ["*"], "output": "*" },
                  { "inputs": ["$context(device).displayName"], "output": "deviceName" },
                  { "inputs": ["$context(device).location"], "output": "location" }
                ]
              }

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

  nodeConnections:
    - from: { name: sensors }
      to: { name: enrich-and-map }
    - from: { name: enrich-and-map }
      to: { name: output }