Ein data flow ist der Pfad, den Daten von der Quelle zum Ziel mit optionalen Transformationen nehmen. Sie können den Datenfluss konfigurieren, indem Sie eine benutzerdefinierte Datenfluss-Ressource erstellen oder die Web-Oberfläche für den Bedienbetrieb verwenden. Eine data flow besteht aus drei Teilen: der source, der transformation und der destination.
Um die Quelle und das Ziel zu definieren, müssen Sie die data flow Endpunkte konfigurieren. Die Transformation ist optional und kann Vorgänge wie das Anreichern der Daten, das Filtern der Daten und das Zuordnen der Daten zu einem anderen Feld umfassen.
Wichtig
Jeder Datenfluss muss den lokalen MQTT-Broker-Standardendpunkt von Azure IoT Einsatz entweder als Quelle oder als Ziel verwenden.
Sie können die Betriebserfahrung in Azure IoT Einsatz verwenden, um eine data flow zu erstellen. Die Betriebserfahrung bietet eine visuelle Schnittstelle zum Konfigurieren des Datenflusses. Sie können auch Bicep verwenden, um eine data flow mithilfe einer Bicep-Datei zu erstellen, oder Kubernetes verwenden, um eine data flow mithilfe einer YAML-Datei zu erstellen.
Lesen Sie weiter, um zu erfahren, wie Sie Quelle, Transformation und Ziel konfigurieren.
Voraussetzungen
Sie können Datenflüsse bereitstellen, sobald Sie über eine Instanz von Azure IoT Einsatz verfügen, indem Sie das Standardprofil und den Endpunkt data flow verwenden. Möglicherweise möchten Sie jedoch Datenfluss-Profile und Endpunkte konfigurieren, um den Datenfluss anzupassen.
Datenfluss-Profil
Wenn Sie keine anderen Skalierungseinstellungen für Ihre Datenflüsse benötigen, verwenden Sie das default-data flow Profil von Azure IoT Einsatz bereitgestellt. Vermeiden Sie das Zuordnen zu vieler Datenflüsse zu einem einzelnen data flow Profil. Wenn Sie über eine große Anzahl von Datenflüssen verfügen, verteilen Sie diese über mehrere Datenflussprofile, um das Risiko zu verringern, dass der Grenzwert für die Konfigurationsgröße des Datenflussprofils von 70 überschritten wird.
Informationen zum Konfigurieren eines neuen data flow-Profils finden Sie unter Configure data flow Profile.
Datenflussendpunkte
Sie benötigen Datenfluss-Endpunkte, um die Quelle und das Ziel des Datenflusses zu konfigurieren. Um schnell loszulegen, verwenden Sie den Standard-Datenfluss-Endpunkt für den lokalen MQTT-Broker. Sie können auch andere Arten von data flow Endpunkten wie Kafka, Event Hubs, OpenTelemetry oder Azure Data Lake Storage erstellen. Weitere Informationen finden Sie unter Datenflussendpunkte konfigurieren.
Loslegen
Wenn Sie über die Voraussetzungen verfügen, können Sie mit dem Erstellen einer data flow beginnen.
Zum Erstellen eines Datenflusses in Operations Experience wählen Sie Datenfluss>Datenfluss erstellen aus.
Wählen Sie den Platzhalternamen new-data-flow aus, um die data flow Eigenschaften festzulegen. Geben Sie den Namen des data flow ein, und wählen Sie das zu verwendende data flow Profil aus. Standardmäßig ist das Datenflussprofil ausgewählt. Weitere Informationen zu data flow Profilen finden Sie unter Configure data flow Profile.
Wichtig
Sie können das Datenflussprofil nur beim Erstellen eines Datenflusses auswählen. Sie können das data flow Profil nicht ändern, nachdem das data flow erstellt wurde.
Wenn Sie das Profil eines vorhandenen Datenflusses ändern möchten, löschen Sie den ursprünglichen Datenfluss und erstellen Sie einen neuen mit dem neuen Datenflussprofil.
Konfigurieren Sie den Quell-, Transformations- und Zielendpunkt für die data flow, indem Sie die Elemente im data flow Diagramm auswählen.
Verwenden Sie den Befehl az iot ops dataflow apply, um einen Datenfluss zu erstellen oder zu ändern.
az iot ops dataflow apply --resource-group <ResourceGroupName> --instance <AioInstanceName> --profile <DataflowProfileName> --name <DataflowName> --config-file <ConfigFilePathAndName>
Der --config-file Parameter ist der Pfad und Dateiname einer JSON-Konfigurationsdatei, die die Ressourceneigenschaften enthält.
Gehen Sie in diesem Beispiel davon aus, dass eine Konfigurationsdatei data-flow.json mit dem folgenden Inhalt im Startverzeichnis des Benutzers gespeichert ist:
{
"mode": "Enabled",
"operations": [
{
"operationType": "Source",
"sourceSettings": {
// See source configuration section
}
},
{
"operationType": "BuiltInTransformation",
"builtInTransformationSettings": {
// See transformation configuration section
}
},
{
"operationType": "Destination",
"destinationSettings": {
// See destination configuration section
}
}
]
}
Hier ist ein Beispielbefehl zum Erstellen oder Aktualisieren eines data flow mithilfe des Standarddatenflussprofils:
az iot ops dataflow apply --resource-group myResourceGroup --instance myAioInstance --profile default --name data-flow --config-file ~/data-flow.json
Erstellen Sie eine Bicep-.bicep-Datei, um einen Datenfluss zu erstellen. In diesem Beispiel wird die Struktur der data flow dargestellt, die die Quell-, Transformations- und Zielkonfigurationen enthält.
param aioInstanceName string = '<AIO_INSTANCE_NAME>'
param customLocationName string = '<CUSTOM_LOCATION_NAME>'
param dataflowName string = '<DATAFLOW_NAME>'
resource aioInstance 'Microsoft.IoTOperations/instances@2026-03-01' existing = {
name: aioInstanceName
}
resource customLocation 'Microsoft.ExtendedLocation/customLocations@2021-08-31-preview' existing = {
name: customLocationName
}
resource defaultDataflowEndpoint 'Microsoft.IoTOperations/instances/dataflowEndpoints@2026-03-01' existing = {
parent: aioInstance
name: 'default'
}
// Pointer to the default data flow profile
resource defaultDataflowProfile 'Microsoft.IoTOperations/instances/dataflowProfiles@2026-03-01' existing = {
parent: aioInstance
name: 'default'
}
resource dataflow 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflows@2026-03-01' = {
// Reference to the parent data flow profile, the default profile in this case
// Same usage as profileRef in Kubernetes YAML
parent: defaultDataflowProfile
name: dataflowName
extendedLocation: {
name: customLocation.id
type: 'CustomLocation'
}
properties: {
mode: 'Enabled'
operations: [
{
operationType: 'Source'
sourceSettings: {
// See source configuration section
}
}
// Transformation optional
{
operationType: 'BuiltInTransformation'
builtInTransformationSettings: {
// See transformation configuration section
}
}
{
operationType: 'Destination'
destinationSettings: {
// See destination configuration section
}
}
]
}
}
Wichtig
Die Verwendung von Kubernetes-Bereitstellungsmanifesten wird in Produktionsumgebungen nicht unterstützt und sollte nur zum Debuggen und Testen verwendet werden.
Erstellen Sie eine Kubernetes-Manifest .yaml Datei, um mit dem Erstellen eines Datenflusses zu beginnen. In diesem Beispiel wird die Struktur der data flow dargestellt, die die Quell-, Transformations- und Zielkonfigurationen enthält.
apiVersion: connectivity.iotoperations.azure.com/v1
kind: Dataflow
metadata:
name: <DATAFLOW_NAME>
namespace: azure-iot-operations
spec:
# Reference to the default data flow profile
# This field is required when configuring via Kubernetes YAML
# The syntax is different when using Bicep
profileRef: default
mode: Enabled
operations:
- operationType: Source
sourceSettings:
# See source configuration section
# Transformation optional
- operationType: BuiltInTransformation
builtInTransformationSettings:
# See transformation configuration section
- operationType: Destination
destinationSettings:
# See destination configuration section
In den folgenden Abschnitten erfahren Sie, wie Sie die Vorgangstypen des data flow konfigurieren.
`Source`
Konfigurieren Sie den Quellendpunkt und Datenquellen (Themen) für den Datenfluss. Sie können den standardmäßigen MQTT-Broker, ein Objekt oder einen benutzerdefinierten MQTT- oder Kafka-Endpunkt als Quelle verwenden.
Vollständige Konfigurationsdetails, einschließlich MQTT-Themen-Wildcards, freigegebene Abonnements, Kafka-Themen und Quellschema, finden Sie unter Konfigurieren einer Datenflussquelle.
Wenn Sie den Standardendpunkt nicht als Quelle verwenden, müssen Sie ihn als Ziel verwenden. Weitere Informationen zur Verwendung des lokalen MQTT-Broker-Endpunkts finden Sie unter "Datenflüsse müssen den lokalen MQTT-Brokerendpunkt verwenden".
Anfordern der Datenträgerpersistenz
Die Datenträgerpersistenz behält den Zustand der Datenflussverarbeitung über Neustarts hinweg bei. Konfigurationsdetails finden Sie unter Konfigurieren der Datenträgerpersistenz.
Der Transformationsvorgang ist der Ort, an dem Sie die Daten aus der Quelle transformieren, bevor Sie sie an das Ziel senden. Transformationen sind optional. Wenn Sie keine Änderungen an den Daten vornehmen müssen, schließen Sie die Transformationsoperation nicht in die Data-Flow-Konfiguration ein. Mehrere Transformationen werden in Phasen miteinander verkettet, unabhängig von der Reihenfolge, in der Sie sie in der Konfiguration angeben. Die Reihenfolge der Phasen lautet immer wie folgt:
-
Anreichern: Fügen Sie weitere Daten zu den Quelldaten hinzu, die einem Dataset und einer Bedingung zugeordnet werden sollen.
-
Filtern: Filtern Sie die Daten anhand einer Bedingung.
-
Zuordnen, Berechnen, Umbenennen oder Hinzufügen einer neuen Eigenschaft: Verschieben Sie Daten aus einem Feld in ein anderes mit optionaler Konvertierung.
Dieser Abschnitt enthält eine Einführung in data flow Transformationen. Ausführlichere Informationen finden Sie unter Zuordnen von Daten mithilfe von Datenflüssen und Anreichern von Datenflüssen.
Wählen Sie in der Betriebsumgebung Datenfluss>Transformation hinzufügen (optional) aus.
{
"operationType": "BuiltInTransformation",
"builtInTransformationSettings": {
"datasets": [
// See section on enriching data
],
"filter": [
// See section on filtering data
],
"map": [
// See section on mapping data
]
}
}
builtInTransformationSettings: {
datasets: [
// See section on enriching data
]
filter: [
// See section on filtering data
]
map: [
// See section on mapping data
]
}
Wichtig
Die Verwendung von Kubernetes-Bereitstellungsmanifesten wird in Produktionsumgebungen nicht unterstützt und sollte nur zum Debuggen und Testen verwendet werden.
builtInTransformationSettings:
datasets:
# See section on enriching data
filter:
# See section on filtering data
map:
# See section on mapping data
Anreichern: Hinzufügen von Verweisdaten
Um die Daten zu anreichern, fügen Sie zuerst im Azure IoT Einsatz statusspeicher ein Verweis-Dataset hinzu. Das Dataset fügt den Quelldaten zusätzliche Daten basierend auf einer Bedingung hinzu. Die Bedingung wird als Feld in den Quelldaten angegeben, die einem Feld im Dataset entsprechen.
Sie können Beispieldaten mithilfe des state-Speichers CLI in den Zustandsspeicher laden. Schlüsselnamen im Statusspeicher entsprechen einem Dataset in der data flow-Konfiguration.
Derzeit wird die Phase Anreichern auf der Betriebserfahrung nicht unterstützt.
Verwenden Sie die Eigenschaft builtInTransformationSettings in der Data-Flow-Konfiguration, um die Daten anzureichern. Verwenden Sie die datasets Eigenschaft, um die Datasets für die Anreicherung anzugeben.
{
"operationType": "BuiltInTransformation",
"builtInTransformationSettings": {
"datasets": [
{
"key": "<DATASET_KEY>",
"inputs": [
"$source.<SOURCE_FIELD>" // ---------------- $1
"$context(<DATASET_KEY>).<DATASET_FIELD>" // - $2
],
"expression": "$1 == $2"
}
]
}
}
In diesem Beispiel wird gezeigt, wie Sie das deviceId Feld in den Quelldaten verwenden können, um mit dem asset Feld im Dataset übereinzustimmen:
builtInTransformationSettings: {
datasets: [
{
key: 'assetDataset'
inputs: [
'$source.deviceId' // ---------------- $1
'$context(assetDataset).asset' // ---- $2
]
expression: '$1 == $2'
}
]
}
Wichtig
Die Verwendung von Kubernetes-Bereitstellungsmanifesten wird in Produktionsumgebungen nicht unterstützt und sollte nur zum Debuggen und Testen verwendet werden.
Sie können z. B. das Feld deviceId in den Quelldaten verwenden, damit es dem Feld asset im Dataset entspricht:
builtInTransformationSettings:
datasets:
- key: assetDataset
inputs:
- $source.deviceId # ------------- $1
- $context(assetDataset).asset # - $2
expression: $1 == $2
Wenn das Dataset über einen Datensatz mit dem Feld asset verfügt, ähnelt es Folgendem:
{
"asset": "thermostat1",
"location": "room1",
"manufacturer": "Contoso"
}
Die Daten aus der Quelle mit dem deviceId-Feld, das thermostat1 entspricht, verfügen über die Felder location und manufacturer in Filter- und Zuordnungsphasen.
Weitere Informationen zur Bedingungssyntax finden Sie unter Anreichern von Daten mithilfe von Datenflüssen
Filter: Filtern von Daten basierend auf einer Bedingung
Verwenden Sie die Filterstufe, um Nachrichten abzulegen, die keine Bedingung erfüllen. Sie können mehrere Filterregeln mit Eingabefeldern und booleschen Ausdrücken definieren.
Vollständige Konfigurationsdetails und Beispiele finden Sie unter Filtern von Daten in einem Datenfluss.
Zuordnung: Verschieben von Daten aus einem Feld in ein anderes
Verwenden Sie den map Vorgang, um die Daten einem anderen Feld mit optionaler Konvertierung zuzuordnen. Geben Sie die Konvertierung als Formel an, die die Felder in den Quelldaten verwendet.
In der Betriebsumgebung können Sie derzeit Daten mithilfe von Compute-, Rename- und Neue Eigenschaft-Transformationen zuordnen.
Berechnen
Verwenden Sie die Computetransformation , um eine Formel auf die Quelldaten anzuwenden. Dieser Vorgang wendet eine Formel auf die Quelldaten an und speichert das Ergebnis in einem Feld.
Wählen Sie unter Transformieren (optional) die Option Compute>Hinzufügen aus.
Geben Sie die erforderlichen Einstellungen ein.
| Einstellung |
Beschreibung |
| Auswählen einer Formel |
Wählen Sie eine vorhandene Formel aus der Dropdownliste aus, oder wählen Sie Benutzerdefiniert aus, um eine Formel manuell einzugeben. |
| Ausgabe |
Geben Sie den Anzeigenamen für die Ausgabe des Ergebnisses an. |
| Formel |
Geben Sie die Formel ein, die auf die Quelldaten angewendet werden soll. |
| Beschreibung |
Geben Sie eine Beschreibung für die Transformation an. |
| Letzter bekannter Wert |
Verwenden Sie optional den letzten bekannten Wert, wenn der aktuelle Wert nicht verfügbar ist. |
Geben Sie eine Formel in das Feld "Formel " ein, oder bearbeiten Sie sie. Die Formel kann die Felder in den Quelldaten verwenden. Geben Sie @ ein, oder drücken Sie STRG+LEERTASTE, um Datenpunkte aus einer Dropdownliste auszuwählen. Wählen Sie für integrierte Formeln den Platzhalter <dataflow> aus, um die Liste der verfügbaren Datenpunkte anzuzeigen.
Geben Sie die MQTT-Metadateneigenschaften im Format @$metadata.user_properties.<property> oder @$metadata.topic ein. Geben Sie $metadata Kopfzeilen mithilfe des Formats @$metadata.<header>ein. Die $metadata-Syntax ist nur für MQTT-Eigenschaften erforderlich, die Teil des Nachrichtenkopfs sind. Weitere Informationen finden Sie unter Feldverweise.
Die Formel kann die Felder in den Quelldaten verwenden. Sie können beispielsweise das Feld temperature in den Quelldaten verwenden, um die Temperatur in Celsius zu konvertieren und im Ausgabefeld temperatureCelsius zu speichern.
Wählen Sie Übernehmen.
Umbenennen
Verwenden Sie die Umbenennungstransformation , um einen Datenpunkt umzubenennen. Dieser Vorgang benennt einen Datenpunkt in den Quelldaten in einen neuen Namen um. Verwenden Sie den neuen Namen in den nachfolgenden Stufen des Datenflusses.
Wählen Sie unter Transformieren (optional) die Option Umbenennen>Hinzufügen aus.
Geben Sie die erforderlichen Einstellungen ein.
| Einstellung |
Beschreibung |
| Datenpunkt |
Wählen Sie einen Datenpunkt aus der Dropdownliste aus, oder geben Sie einen $metadata-Header ein. |
| Neuer Datenpunktname |
Geben Sie den neuen Namen für den Datenpunkt ein. |
| Beschreibung |
Geben Sie eine Beschreibung für die Transformation an. |
Geben Sie die MQTT-Metadateneigenschaften im Format @$metadata.user_properties.<property> oder @$metadata.topic ein. Geben Sie $metadata Kopfzeilen mithilfe des Formats @$metadata.<header>ein. Die $metadata-Syntax ist nur für MQTT-Eigenschaften erforderlich, die Teil des Nachrichtenkopfs sind. Weitere Informationen finden Sie unter Feldverweise.
Wählen Sie Übernehmen.
Neue Eigenschaft
Verwenden Sie die Neue-Eigenschaft-Transformation, um den Quelldaten eine neue Eigenschaft hinzuzufügen. Dieser Vorgang fügt der Quelldaten eine neue Eigenschaft hinzu. Verwenden Sie die neue Eigenschaft in den nachfolgenden Phasen des Datenflusses.
Wählen Sie unter Transformieren (optional) die Option Neue Eigenschaft>Hinzufügen aus.
Geben Sie die erforderlichen Einstellungen ein.
| Einstellung |
Beschreibung |
| Eigenschaftsschlüssel |
Geben Sie den Schlüssel für die neue Eigenschaft ein. |
| Eigenschaftswert |
Geben Sie den Wert für die neue Eigenschaft ein. |
| Beschreibung |
Geben Sie eine Beschreibung für die neue Eigenschaft an. |
Wählen Sie Übernehmen.
Sie können z. B. das temperature Feld in den Quelldaten verwenden, um die Temperatur in Celsius zu konvertieren und im Feld zu temperatureCelsius speichern. Erweitern Sie die Quelldaten mit dem location Feld aus dem Kontext-Dataset:
{
"operationType": "BuiltInTransformation",
"builtInTransformationSettings": {
"map": [
{
"inputs": [
"$source.temperature ? $last" // ---------------- $1
],
"output": "temperatureCelsius",
"expression": "($1 - 32) * 5/9"
},
{
"inputs": [
"$context(assetDataset).location" // - $2
],
"output": "location"
}
]
}
}
Sie können auf die Eigenschaften der MQTT-Metadaten zugreifen, indem Sie das Format $metadata.user_properties.<property> oder $metadata.topic verwenden. Sie können auch $metadata Kopfzeilen eingeben, indem Sie das Format $metadata.<header>verwenden. Weitere Informationen finden Sie unter Feldverweise.
Sie können z. B. das temperature Feld in den Quelldaten verwenden, um die Temperatur in Celsius zu konvertieren und im Feld zu temperatureCelsius speichern. Erweitern Sie die Quelldaten mit dem location Feld aus dem Kontext-Dataset:
builtInTransformationSettings: {
map: [
{
inputs: [
'temperature'
]
output: 'temperatureCelsius'
expression: '($1 - 32) * 5/9'
}
{
inputs: [
'$context(assetDataset).location'
]
output: 'location'
}
]
}
Wichtig
Die Verwendung von Kubernetes-Bereitstellungsmanifesten wird in Produktionsumgebungen nicht unterstützt und sollte nur zum Debuggen und Testen verwendet werden.
Sie können auf die Eigenschaften der MQTT-Metadaten zugreifen, indem Sie das Format $metadata.user_properties.<property> oder $metadata.topic verwenden. Sie können auch $metadata Kopfzeilen eingeben, indem Sie das Format $metadata.<header>verwenden. Weitere Informationen finden Sie unter Feldverweise.
Sie können z. B. das temperature Feld in den Quelldaten verwenden, um die Temperatur in Celsius zu konvertieren und im Feld zu temperatureCelsius speichern. Erweitern Sie die Quelldaten mit dem location Feld aus dem Kontext-Dataset:
builtInTransformationSettings:
map:
- inputs:
- temperature # - $1
expression: "($1 - 32) * 5/9"
output: temperatureCelsius
- inputs:
- $context(assetDataset).location
output: location
Weitere Informationen finden Sie unter Zuordnen von Daten mithilfe von Datenflüssen.
Entfernen
Standardmäßig enthält das Ausgabeschema alle Datenpunkte. Entfernen Sie alle Datenpunkte aus dem Ziel, indem Sie die Remove-Transformation verwenden.
Wählen Sie unter "Transformieren (optional)" die Option "Entfernen" aus.
Wählen Sie den zu entfernenden Datenpunkt aus dem Ausgabeschema aus.
Wählen Sie Übernehmen.
Um einen Datenpunkt aus dem Ausgabeschema zu entfernen, verwenden Sie die Eigenschaft builtInTransformationSettings in der Datenflusskonfiguration. Verwenden Sie die map Eigenschaft, um die zu entfernenden Datenpunkte anzugeben.
{
"operationType": "BuiltInTransformation",
"builtInTransformationSettings": {
"map": [
{
"inputs": [
"*"
],
"output": "*"
},
{
"inputs": [
"weight"
],
"output": ""
}
{
"inputs": [
"weight.SourceTimestamp"
],
"output": ""
},
{
"inputs": [
"weight.Value"
],
"output": ""
},
{
"inputs": [
"weight.StatusCode"
],
"output": ""
},
{
"inputs": [
"weight.StatusCode.Code"
],
"output": ""
},
{
"inputs": [
"weight.StatusCode.Symbol"
],
"output": ""
}
]
}
}
builtInTransformationSettings: {
map: [
{
inputs: [
'*'
]
output: '*'
}
{
inputs: [
'weight'
]
output: ''
}
{
inputs: [
'weight.SourceTimestamp'
]
output: ''
}
{
inputs: [
'weight.Value'
]
output: ''
}
{
inputs: [
'weight.StatusCode'
]
output: ''
}
{
inputs: [
'weight.StatusCode.Code'
]
output: ''
}
{
inputs: [
'weight.StatusCode.Symbol'
]
output: ''
}
]
}
Wichtig
Die Verwendung von Kubernetes-Bereitstellungsmanifesten wird in Produktionsumgebungen nicht unterstützt und sollte nur zum Debuggen und Testen verwendet werden.
builtInTransformationSettings:
map:
- type: PassThrough
inputs:
- "*"
output: "*"
- inputs:
- weight
output: ""
- inputs:
- weight.SourceTimestamp
output: ""
- inputs:
- weight.Value
output: ""
- inputs:
- weight.StatusCode
output: ""
- inputs:
- weight.StatusCode.Code
output: ""
- inputs:
- weight.StatusCode.Symbol
output: ""
Weitere Informationen finden Sie unter Zuordnen von Daten mithilfe von Datenflüssen.
Serialisieren von Daten nach einem Schema
Wenn Sie die Daten vor dem Senden an das Ziel serialisieren möchten, geben Sie ein Schema- und Serialisierungsformat an. Ausführliche Informationen finden Sie unter Serialisieren der Ausgabe mit einem Schema.
Bestimmungsort
Konfigurieren Sie den Zielendpunkt und das Datenziel (Thema, Container oder Tabelle) für den Datenfluss. Sie können jeden unterstützten Endpunkttyp als Ziel verwenden, einschließlich MQTT, Kafka, Azure Data Lake Storage, Microsoft Fabric, Azure Data Explorer und lokaler Speicher.
Vollständige Konfigurationsdetails, einschließlich der Datenzieltabelle, dynamische Zielthemen und Ausgabe serialisierung, finden Sie unter Konfigurieren eines Datenflussziels.
Um Daten an ein anderes Ziel als den lokalen MQTT-Broker zu senden, erstellen Sie einen data flow Endpunkt. Informationen dazu finden Sie unter Datenflussendpunkte konfigurieren.
Wichtig
Storage Endpunkte benötigen ein schema für die Serialisierung. Um data flow mit Microsoft Fabric OneLake, Azure Data Lake Storage, Azure Data Explorer oder lokalen Storage zu verwenden, müssen Sie einen Schemaverweis spezifizieren.
Beispiel
Das folgende Beispiel ist eine data flow Konfiguration, die den MQTT-Endpunkt für die Quelle und das Ziel verwendet. Die Quelle filtert die Daten aus dem MQTT-Thema azure-iot-operations/data/thermostat. Die Transformation konvertiert die Temperatur in Fahrenheit und filtert die Daten, bei denen das Verhältnis Temperatur multipliziert mit Luftfeuchtigkeit kleiner als 100.000 ist. Das Ziel sendet die Daten an das MQTT-Thema factory.
Verwenden Sie den Befehl az iot ops dataflow apply, um einen Datenfluss zu erstellen oder zu ändern.
az iot ops dataflow apply --resource-group <ResourceGroupName> --instance <AioInstanceName> --profile <DataflowProfileName> --name <DataflowName> --config-file <ConfigFilePathAndName>
Der --config-file Parameter ist der Pfad und Dateiname einer JSON-Konfigurationsdatei, die die Ressourceneigenschaften enthält.
Gehen Sie in diesem Beispiel davon aus, dass eine Konfigurationsdatei data-flow.json mit dem folgenden Inhalt im Startverzeichnis des Benutzers gespeichert ist:
{
"mode": "Enabled",
"operations": [
{
"operationType": "Source",
"sourceSettings": {
"dataSources": [
"thermostats/+/sensor/temperature/#",
"humidifiers/+/sensor/humidity/#"
],
"endpointRef": "default",
"serializationFormat": "Json"
}
},
{
"builtInTransformationSettings": {
"datasets": [],
"filter": [
{
"expression": "$1 * $2 < 100000",
"inputs": [
"temperature.Value",
"\"Tag 10\".Value"
],
"type": "Filter"
}
],
"map": [
{
"inputs": [
"*"
],
"output": "*",
"type": "PassThrough"
},
{
"expression": "cToF($1)",
"inputs": [
"temperature.Value"
],
"output": "TemperatureF",
"type": "Compute"
},
{
"inputs": [
"@\"Tag 10\".Value"
],
"output": "Humidity",
"type": "Rename"
}
],
"serializationFormat": "Json"
},
"operationType": "BuiltInTransformation"
},
{
"destinationSettings": {
"dataDestination": "factory",
"endpointRef": "default"
},
"operationType": "Destination"
}
]
}
Hier ist ein Beispielbefehl zum Erstellen oder Aktualisieren eines data flow mithilfe des Standarddatenflussprofils:
az iot ops dataflow apply --resource-group myResourceGroup --instance myAioInstance --profile default --name data-flow --config-file ~/data-flow.json
Hier ist ein weiteres Beispiel für die dynamische Themenübersetzung, um Nachrichten von verschiedenen Thermometern an gerätespezifische Themen weiterzuleiten:
{
"mode": "Enabled",
"operations": [
{
"operationType": "Source",
"sourceSettings": {
"dataSources": [
"thermostats/+/sensor/temperature"
],
"endpointRef": "default",
"serializationFormat": "Json"
}
},
{
"destinationSettings": {
"dataDestination": "processed/device/${inputTopic.2}/temperature",
"endpointRef": "default"
},
"operationType": "Destination"
}
]
}
Diese Konfiguration verarbeitet Nachrichten von thermostats/device1/sensor/temperature und sendet sie an processed/device/device1/temperature.
param aioInstanceName string = '<AIO_INSTANCE_NAME>'
param customLocationName string = '<CUSTOM_LOCATION_NAME>'
param dataflowName string = '<DATAFLOW_NAME>'
resource aioInstance 'Microsoft.IoTOperations/instances@2026-03-01' existing = {
name: aioInstanceName
}
resource customLocation 'Microsoft.ExtendedLocation/customLocations@2021-08-31-preview' existing = {
name: customLocationName
}
// Pointer to the default data flow endpoint
resource defaultDataflowEndpoint 'Microsoft.IoTOperations/instances/dataflowEndpoints@2026-03-01' existing = {
parent: aioInstance
name: 'default'
}
// Pointer to the default data flow profile
resource defaultDataflowProfile 'Microsoft.IoTOperations/instances/dataflowProfiles@2026-03-01' existing = {
parent: aioInstance
name: 'default'
}
resource dataflow 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflows@2026-03-01' = {
// Reference to the parent data flow profile, the default profile in this case
// Same usage as profileRef in Kubernetes YAML
parent: defaultDataflowProfile
name: dataflowName
extendedLocation: {
name: customLocation.id
type: 'CustomLocation'
}
properties: {
mode: 'Enabled'
operations: [
{
operationType: 'Source'
sourceSettings: {
// Use the default MQTT endpoint as the source
endpointRef: defaultDataflowEndpoint.name
// Filter the data from the MQTT topic azure-iot-operations/data/thermostat
dataSources: [
'azure-iot-operations/data/thermostat'
]
}
}
// Transformation optional
{
operationType: 'BuiltInTransformation'
builtInTransformationSettings: {
// Filter the data where temperature * "Tag 10" < 100000
filter: [
{
inputs: [
'temperature.Value'
'"Tag 10".Value'
]
expression: '$1 * $2 < 100000'
}
]
map: [
// Passthrough all values by default
{
inputs: [
'*'
]
output: '*'
}
// Convert temperature to Fahrenheit and output it to TemperatureF
{
inputs: [
'temperature.Value'
]
output: 'TemperatureF'
expression: 'cToF($1)'
}
// Extract the "Tag 10" value and output it to Humidity
{
inputs: [
'"Tag 10".Value'
]
output: 'Humidity'
}
]
}
}
{
operationType: 'Destination'
destinationSettings: {
// Use the default MQTT endpoint as the destination
endpointRef: defaultDataflowEndpoint.name
// Send the data to the MQTT topic factory
dataDestination: 'factory'
}
}
]
}
}
Wichtig
Die Verwendung von Kubernetes-Bereitstellungsmanifesten wird in Produktionsumgebungen nicht unterstützt und sollte nur zum Debuggen und Testen verwendet werden.
apiVersion: connectivity.iotoperations.azure.com/v1
kind: Dataflow
metadata:
name: my-dataflow
namespace: azure-iot-operations
spec:
# Reference to the default data flow profile
profileRef: default
mode: Enabled
operations:
- operationType: Source
sourceSettings:
# Use the default MQTT endpoint as the source
endpointRef: default
# Filter the data from the MQTT topic azure-iot-operations/data/thermostat
dataSources:
- azure-iot-operations/data/thermostat
# Transformation optional
- operationType: builtInTransformation
builtInTransformationSettings:
# Filter the data where temperature * "Tag 10" < 100000
filter:
- inputs:
- 'temperature.Value'
- '"Tag 10".Value'
expression: '$1 * $2 < 100000'
map:
# Passthrough all values by default
- inputs:
- '*'
output: '*'
# Convert temperature to Fahrenheit and output it to TemperatureF
- inputs:
- temperature.Value
output: TemperatureF
expression: cToF($1)
# Extract the "Tag 10" value and output it to Humidity
- inputs:
- '"Tag 10".Value'
output: 'Humidity'
- operationType: Destination
destinationSettings:
# Use the default MQTT endpoint as the destination
endpointRef: default
# Send the data to the MQTT topic factory
dataDestination: factory
Weitere Beispiele für data flow Konfigurationen finden Sie unter Azure REST-API – Data flow und quickstart Bicep.
Überprüfen, ob ein data flow funktioniert
Um zu überprüfen, ob die data flow funktioniert, folgen Sie Tutorial: Bidirektionale MQTT-Brücke zu Azure Event Grid.
Exportieren der Datenflusskonfiguration
Um die Datenfluss-Konfiguration zu exportieren, verwenden Sie die Betriebsoberfläche oder exportieren Sie die benutzerdefinierte Datenflussressource.
Wählen Sie die data flow aus, die Sie exportieren möchten, und wählen Sie dann auf der Symbolleiste Export aus.
Verwenden Sie den Befehl az iot ops dataflow show, um eine data flow zu exportieren.
az iot ops dataflow show --resource-group <ResourceGroupName> --instance <AioInstanceName> --name <DataflowName> --profile <DataflowProfileName> --output json > my-dataflow.json
Hier ist ein Beispielbefehl zum Exportieren eines data flow namens data-flow in eine JSON-Datei mit dem Namen data-flow.json:
az iot ops dataflow show --resource-group myResourceGroup --instance myAioInstance --profile default --name data-flow --output json > data-flow.json
Bicep nutzt den Infrastructure-as-Code-Ansatz, und es ist kein Export erforderlich. Verwenden Sie die Datei Bicep, um eine data flow zu erstellen, um Datenflüsse schnell einzurichten und zu konfigurieren.
Wichtig
Die Verwendung von Kubernetes-Bereitstellungsmanifesten wird in Produktionsumgebungen nicht unterstützt und sollte nur zum Debuggen und Testen verwendet werden.
kubectl get dataflow my-dataflow -o yaml > my-dataflow.yaml
Richtige Datenflusskonfiguration
Um sicherzustellen, dass die data flow wie erwartet funktioniert, überprüfen Sie die folgenden Bedingungen:
- Der standardmäßige MQTT-data flow-Endpunkt must als Quelle oder Ziel verwendet werden.
- Das Dataflow-Profil ist vorhanden und wird in der Datenflusskonfiguration referenziert.
- Die Quelle ist entweder ein MQTT-Endpunkt, ein Kafka-Endpunkt oder ein Objekt. Speichertyp-Endpunkte können Sie nicht als Quelle verwenden.
- Wenn Sie Event Grid als Quelle verwenden, legen Sie die Anzahl der Datenflussprofilinstanzen auf 1 fest, da der Event Grid MQTT-Broker keine freigegebenen Abonnements unterstützt.
- Wenn Sie Event Hubs als Quelle verwenden, ist jeder Event Hub im Namespace ein separates Kafka-Thema, und Sie müssen jede als Datenquelle angeben.
- Die Transformation wird, sofern verwendet, mit der richtigen Syntax konfiguriert, einschließlich der richtigen Maskierung von Sonderzeichen.
- Wenn Sie Speichertyp-Endpunkte als Ziel verwenden, wird ein Schema angegeben.
- Wenn Sie dynamische Zielthemen für MQTT-Endpunkte verwenden, stellen Sie sicher, dass Themenvariablen auf gültige Segmente verweisen.
Nächste Schritte
