Skapa dataflöden i Azure IoT Operations

1. Om du vill skapa ett dataflöde i driftmiljön väljer du Dataflöde>Skapa dataflöde.

2. Välj platshållarnamnet new-data-flow för att ange dataflödesegenskaperna. Ange namnet på dataflödet och välj den dataflödesprofil som ska användas. Standardprofilen för dataflöde är markerad som standard. Mer information om dataflödesprofiler finns i Konfigurera dataflödesprofil.

   

   Viktigt!

   Du kan bara välja dataflödesprofilen när du skapar ett dataflöde. Du kan inte ändra dataflödesprofilen när dataflödet har skapats. Om du vill ändra dataflödesprofilen för ett befintligt dataflöde tar du bort det ursprungliga dataflödet och skapar ett nytt med den nya dataflödesprofilen.

3. Konfigurera käll-, transformerings- och målslutpunkten för dataflödet genom att välja objekten i dataflödesdiagrammet.

   

Använd kommandot az iot ops dataflow apply för att skapa eller ändra ett dataflöde.

az iot ops dataflow apply --resource-group <ResourceGroupName> --instance <AioInstanceName> --profile <DataflowProfileName> --name <DataflowName> --config-file <ConfigFilePathAndName>

Parametern --config-file är sökvägen och filnamnet för en JSON-konfigurationsfil som innehåller resursegenskaperna.

I det här exemplet antar du att en konfigurationsfil med namnet data-flow.json med följande innehåll lagras i användarens hemkatalog:

{
    "mode": "Enabled",
    "operations": [
        {
            "operationType": "Source",
            "sourceSettings": {
                // See source configuration section
            }
        },
        {
            "operationType": "BuiltInTransformation",
            "builtInTransformationSettings": {
                // See transformation configuration section
            }
        },
        {
            "operationType": "Destination",
            "destinationSettings": {
                // See destination configuration section
            }
        }
    ]
}

Här är ett exempelkommando för att skapa eller uppdatera ett dataflöde med hjälp av standardprofilen för dataflöde:

az iot ops dataflow apply --resource-group myResourceGroup --instance myAioInstance --profile default --name data-flow --config-file ~/data-flow.json

Skapa en Bicep-fil .bicep för att börja skapa ett dataflöde. Det här exemplet visar strukturen för dataflödet som innehåller käll-, transformerings- och målkonfigurationerna.

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

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

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

resource defaultDataflowEndpoint 'Microsoft.IoTOperations/instances/dataflowEndpoints@2024-11-01' existing = {
  parent: aioInstance
  name: 'default'
}

// Pointer to the default data flow profile
resource defaultDataflowProfile 'Microsoft.IoTOperations/instances/dataflowProfiles@2024-11-01' existing = {
  parent: aioInstance
  name: 'default'
}

resource dataflow 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflows@2024-11-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
        }
      }
    ]
  }
}

Skapa en Kubernetes-manifestfil .yaml för att börja skapa ett dataflöde. Det här exemplet visar strukturen för dataflödet som innehåller käll-, transformerings- och målkonfigurationerna.

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

1. Under Källinformation väljer du Meddelandehanterare.

   

2. Ange följande inställningar för meddelandemäklarens källa:

   Inställning	beskrivning
   Slutpunkt för dataflöde	Välj standard för att använda standard MQTT-meddelandemäklarens slutpunkt.
   Ämne	Ämnesfiltret som du vill prenumerera på för inkommande meddelanden. Använd Ämne(er)>Lägg till rad för att lägga till flera ämnen. Mer information om ämnen finns i Konfigurera MQTT- eller Kafka-ämnen.
   Meddelandeschema	Schemat som ska användas för att deserialisera inkommande meddelanden. Se Ange schema för att deserialisera data.

3. Välj Använd.

Här är ett exempel på en källslutpunktskonfiguration för standard-MQTT-brokers slutpunkt:

{
  "operationType": "Source",
  "sourceSettings": {
    "endpointRef": "default",
    "dataSources": [
      "thermostats/+/sensor/temperature/#",
      "humidifiers/+/sensor/humidity/#"
    ],
    "endpointRef": "default"
  }
}

Slutpunkten för meddelandeförmedlare är konfigurerad i Bicep-filen. Följande slutpunkt är till exempel en källa för dataflödet.

sourceSettings: {
  endpointRef: 'default'
  dataSources: [
    'thermostats/+/sensor/temperature/#'
    'humidifiers/+/sensor/humidity/#'
  ]
}

Om du till exempel vill konfigurera en källa med hjälp av en slutpunkt för meddelandekö och två ämnesfilter använder du följande konfiguration:

sourceSettings:
  endpointRef: default
  dataSources:
    - thermostats/+/sensor/temperature/#
    - humidifiers/+/sensor/humidity/#

Du kan använda en tillgång som källa för dataflödet. Användning av en tillgång som källa är endast tillgängligt i driftupplevelsen.

1. Under Källinformation väljer du Tillgång.

2. Välj den tillgång som du vill använda som källslutpunkt.

3. Välj Fortsätt.

   En lista över datapunkter för den valda tillgången visas.

   

4. Välj Använd för att använda tillgången som källslutpunkt.

Det är endast möjligt att konfigurera en tillgång som källa i operativ miljö.

Det är endast möjligt att konfigurera en tillgång som källa i operativ miljö.

Det är endast möjligt att konfigurera en tillgång som källa i operativ miljö.

1. Under Källinformation väljer du Meddelandehanterare.

   

2. Ange följande inställningar för meddelandemäklarens källa:

   Inställning	beskrivning
   Slutpunkt för dataflöde	Använd knappen Välj om för att välja en anpassad MQTT- eller Kafka-dataflödesslutpunkt. Mer information finns i Konfigurera MQTT-dataflödesslutpunkter eller Konfigurera Azure Event Hubs- och Kafka-dataflödesslutpunkter.
   Ämne	Ämnesfiltret som du vill prenumerera på för inkommande meddelanden. Använd Ämne(er)>Lägg till rad för att lägga till flera ämnen. Mer information om ämnen finns i Konfigurera MQTT- eller Kafka-ämnen.
   Meddelandeschema	Schemat som ska användas för att deserialisera inkommande meddelanden. Se Ange schema för att deserialisera data.

3. Välj Använd.

Ersätt platshållarvärden med ditt anpassade slutpunktsnamn och ämnen.

{
    "operationType": "Source",
    "sourceSettings": {
        "endpointRef": "<CUSTOM_ENDPOINT_NAME>",
        "dataSources": [
            "<TOPIC_1>",
            "<TOPIC_2>"
        ]
    }
}

Ersätt platshållarvärden med ditt anpassade slutpunktsnamn och ämnen.

sourceSettings: {
  endpointRef: '<CUSTOM_ENDPOINT_NAME>'
  dataSources: [
    '<TOPIC_1>'
    '<TOPIC_2>'
    // See section on configuring MQTT or Kafka topics for more information
  ]
}

Ersätt platshållarvärden med ditt anpassade slutpunktsnamn och ämnen.

sourceSettings:
  endpointRef: <CUSTOM_ENDPOINT_NAME>
  dataSources:
  - <TOPIC_1>
  - <TOPIC_2>
  # See section on configuring MQTT or Kafka topics for more information

I dataflödesinformationen för operativ erfarenhet väljMeddelandehanterare och använd sedan fältet Topic(s) för att ange MQTT-topicfiltren som ska prenumerera på inkommande meddelanden. Du kan lägga till flera MQTT-ämnen genom att välja Lägg till rad och ange ett nytt ämne.

{
    "operationType": "Source",
    "sourceSettings": {
        "endpointRef": "<MESSAGE_BROKER_ENDPOINT_NAME>",
        "dataSources": [
            "<TOPIC_FILTER_1>",
            "<TOPIC_FILTER_2>"
            // Add more topic filters as needed
        ]
    }
}

Exempel med flera MQTT-ämnesfilter med jokertecken:

{
    "operationType": "Source",
    "sourceSettings": {
        "endpointRef": "default",
        "dataSources": [
            "thermostats/+/sensor/temperature/#",
            "humidifiers/+/sensor/humidity/#"
        ]
    }
}

Här används jokertecknet + för att välja alla enheter under avsnitten thermostats och humidifiers . Jokertecknet # används för att välja alla sensormeddelanden under alla underämnen i ämnena temperature och humidity .

sourceSettings: {
  endpointRef: '<MESSAGE_BROKER_ENDPOINT_NAME>'
  dataSources: [
    '<TOPIC_FILTER_1>'
    '<TOPIC_FILTER_2>'
    // Add more topic filters as needed
  ]
}

Exempel med flera MQTT-ämnesfilter med jokertecken:

sourceSettings: {
  endpointRef: 'default'
  dataSources: [
    'thermostats/+/sensor/temperature/#'
    'humidifiers/+/sensor/humidity/#'
  ]
}

Här används jokertecknet + för att välja alla enheter under avsnitten thermostats och humidifiers . Jokertecknet # används för att välja alla sensormeddelanden under alla underämnen i ämnena temperature och humidity .

sourceSettings:
  endpointRef: <ENDPOINT_NAME>
  dataSources:
    - <TOPIC_FILTER_1>
    - <TOPIC_FILTER_2>
    # Add more topic filters as needed

Exempel med flera ämnesfilter med jokertecken:

sourceSettings:
  endpointRef: default
  dataSources:
    - thermostats/+/sensor/temperature/#
    - humidifiers/+/sensor/humidity/#

Här används jokertecknet + för att välja alla enheter under avsnitten thermostats och humidifiers . Jokertecknet # används för att markera alla meddelanden under alla underämnen i ämnena temperature och humidity .

I dataflödet för operativ erfarenhet väljer du Meddelandehanterare och använder fältet Ämne för att ange den delade prenumerationsgruppen och ämnet.

{
  "operationType": "Source",
  "sourceSettings": {
    "dataSources": [
      "$shared/<GROUP_NAME>/<TOPIC_FILTER>"
    ]
  }
}

sourceSettings: {
  dataSources: [
    '$shared/<GROUP_NAME>/<TOPIC_FILTER>'
  ]
}

sourceSettings:
  dataSources:
    - $shared/<GROUP_NAME>/<TOPIC_FILTER>

I dataflödet för driftserfarenheter Source details väljer du Meddelandehanterare, använd sedan fältet Ämne för att ange Kafka-ämnesfiltret som du vill prenumerera på för inkommande meddelanden.

{
  "operationType": "Source",
  "sourceSettings": {
    "endpointRef": "<KAFKA_ENDPOINT_NAME>",
    "dataSources": [
      "<KAFKA_TOPIC_1>",
      "<KAFKA_TOPIC_2>"
      // Add more Kafka topics as needed
    ]
  }
}

sourceSettings: {
  endpointRef: '<KAFKA_ENDPOINT_NAME>'
  dataSources: [
    '<KAFKA_TOPIC_1>'
    '<KAFKA_TOPIC_2>'
    // Add more Kafka topics as needed
  ]
}

sourceSettings:
  endpointRef: <KAFKA_ENDPOINT_NAME>
  dataSources:
    - <KAFKA_TOPIC_1>
    - <KAFKA_TOPIC_2>
    # Add more Kafka topics as needed

I dataflödet för driftserfarenhet väljer du Meddelandemäklare och använder Meddelandeschemafältet för att specificera schemat. Du kan använda knappen Ladda upp för att ladda upp en schemafil först. Mer information finns i Förstå meddelandescheman.

{
  "operationType": "Source",
  "sourceSettings": {
    "endpointRef": "<ENDPOINT_NAME>",
    "serializationFormat": "Json",
    "schemaRef": "aio-sr://<SCHEMA_NAMESPACE>/<SCHEMA_NAME>:<VERSION>"
  }
}

När du använder schemaregistret för att lagra schemat kan du referera till det i dataflödeskonfigurationen.

sourceSettings: {
  serializationFormat: 'Json'
  schemaRef: 'aio-sr://<SCHEMA_NAMESPACE>/<SCHEMA_NAME>:<VERSION>'
}

När du använder schemaregistret för att lagra schemat kan du referera till det i dataflödeskonfigurationen.

sourceSettings:
  serializationFormat: Json
  schemaRef: 'aio-sr://<SCHEMA_NAMESPACE>/<SCHEMA_NAME>:<VERSION>'

När du skapar eller redigerar ett dataflöde väljer du Redigera och sedan Ja bredvid Begär datapersistence.

Lägg till egenskapen i requestDiskPersistence dataflödeskonfigurationsfilen:

{
    "mode": "Enabled",
    "requestDiskPersistence": "Enabled",
    "operations": [
        // ... your data flow operations
    ]
}

Lägg till requestDiskPersistence-egenskapen i din dataflödesresurs. API-versionen är 2025-07-01-preview eller senare:

resource dataflow 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflows@2025-07-01-preview' = {
  parent: defaultDataflowProfile
  name: dataflowName
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    mode: 'Enabled'
    requestDiskPersistence: 'Enabled'
    operations: [
      // ... your data flow operations
    ]
  }
}

Lägg till egenskapen i requestDiskPersistence dataflödesspecifikationen. API-versionen är connectivity.iotoperations.azure.com/v1beta1 eller senare:

apiVersion: connectivity.iotoperations.azure.com/v1beta1
kind: Dataflow
metadata:
  name: <DATAFLOW_NAME>
  namespace: azure-iot-operations
spec:
  profileRef: default 
  mode: Enabled
  requestDiskPersistence: Enabled
  operations:
    # ... your data flow operations

I driftupplevelsen väljer du Dataflöde>Lägg till transformering (valfritt).

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

builtInTransformationSettings:
  datasets:
    # See section on enriching data
  filter:
    # See section on filtering data
  map:
    # See section on mapping data

För närvarande stöds inte Enrich-fasen i driftupplevelsen.

Om du vill utöka data kan du använda builtInTransformationSettings egenskapen i dataflödeskonfigurationen. Egenskapen datasets används för att ange de datauppsättningar som ska användas för berikning.

{
  "operationType": "BuiltInTransformation",
  "builtInTransformationSettings": {
    "datasets": [
      {
        "key": "<DATASET_KEY>",
        "inputs": [
          "$source.<SOURCE_FIELD>" // ---------------- $1
          "$context(<DATASET_KEY>).<DATASET_FIELD>" // - $2
        ],
        "expression": "$1 == $2"
      }
    ]
  }
}

Det här exemplet visar hur du kan använda fältet deviceId i källdata för att matcha asset fältet i datauppsättningen:

builtInTransformationSettings: {
  datasets: [
    {
      key: 'assetDataset'
      inputs: [
        '$source.deviceId' // ---------------- $1
        '$context(assetDataset).asset' // ---- $2
      ]
      expression: '$1 == $2'
    }
  ]
}

Du kan till exempel använda fältet deviceId i källdata för att matcha asset fältet i datauppsättningen:

builtInTransformationSettings:
  datasets:
  - key: assetDataset
    inputs:
      - $source.deviceId # ------------- $1
      - $context(assetDataset).asset # - $2
    expression: $1 == $2

1. Under Transformera (valfritt) väljer du Filtrera>lägg till.

   

2. Ange de inställningar som krävs.

   Inställning	beskrivning
   Filtervillkor	Villkoret för att filtrera data baserat på ett fält i källdata.
   beskrivning	Ange en beskrivning för filtervillkoret.

   I fältet för filtervillkor anger du @ eller väljer Ctrl + Blanksteg för att välja datapunkter från en listruta.

   Du kan ange MQTT-metadataegenskaper med formatet @$metadata.user_properties.<property> eller @$metadata.topic. Du kan också ange $metadata rubriker med formatet @$metadata.<header>. Syntaxen $metadata behövs bara för MQTT-egenskaper som ingår i meddelandehuvudet. Mer information finns i fältreferenser.

   Villkoret kan använda fälten i källdata. Du kan till exempel använda ett filtervillkor som @temperature > 20 att filtrera data som är mindre än eller lika med 20 baserat på temperaturfältet.

3. Välj Använd.

Du kan till exempel använda fältet temperature i källdata för att filtrera data:

{
  "operationType": "BuiltInTransformation",
  "builtInTransformationSettings": {
    "filter": [
      {
        "inputs": [
          "$source.temperature ? $last" // ---------------- $1
        ],
        "expression": "$1 > 20"
      }
    ]
  }
}

Du kan till exempel använda fältet temperature i källdata för att filtrera data:

builtInTransformationSettings: {
  filter: [
    {
      inputs: [
        'temperature ? $last'
      ]
      expression: '$1 > 20'
    }
  ]
}

Om fältet temperature är större än 20 skickas data till nästa steg. Om fältet temperature är mindre än eller lika med 20 filtreras data.

Du kan till exempel använda fältet temperature i källdata för att filtrera data:

builtInTransformationSettings:
  filter:
    - inputs:
      - temperature ? $last # - $1
      expression: "$1 > 20"

Om fältet temperature är större än 20 skickas data till nästa steg. Om fältet temperature är mindre än eller lika med 20 filtreras data.

1. Under Transformera (valfritt) väljer du Ny egenskap>Lägg till.

   

2. Ange de inställningar som krävs.

   Inställning	beskrivning
   Egenskapsnyckel	Ange nyckeln för den nya egenskapen.
   Egenskapsvärde	Ange värdet för den nya egenskapen.
   beskrivning	Ange en beskrivning för den nya egenskapen.

3. Välj Använd.

Du kan till exempel använda fältet temperature i källdata för att konvertera temperaturen till Celsius och lagra den i temperatureCelsius fältet. Du kan också utöka källdata med fältet location från datauppsättningen för kontextualisering:

{
  "operationType": "BuiltInTransformation",
  "builtInTransformationSettings": {
    "map": [
      {
        "inputs": [
          "$source.temperature ? $last" // ---------------- $1
        ],
        "output": "temperatureCelsius",
        "expression": "($1 - 32) * 5/9"
      },
      {
        "inputs": [
          "$context(assetDataset).location" // - $2
        ],
        "output": "location"
      }
    ]
  }
}

Du kan komma åt MQTT-metadataegenskaper med formatet $metadata.user_properties.<property> eller $metadata.topic. Du kan också ange $metadata rubriker med formatet $metadata.<header>. Mer information finns i fältreferenser.

Du kan till exempel använda fältet temperature i källdata för att konvertera temperaturen till Celsius och lagra den i temperatureCelsius fältet. Du kan också utöka källdata med fältet location från datauppsättningen för kontextualisering:

builtInTransformationSettings: {
  map: [
    {
      inputs: [
        'temperature'
      ]
      output: 'temperatureCelsius'
      expression: '($1 - 32) * 5/9'
    }
    {
      inputs: [
        '$context(assetDataset).location'
      ]
      output: 'location'
    }
  ]
}

Du kan komma åt MQTT-metadataegenskaper med formatet $metadata.user_properties.<property> eller $metadata.topic. Du kan också ange $metadata rubriker med formatet $metadata.<header>. Mer information finns i fältreferenser.

Du kan till exempel använda fältet temperature i källdata för att konvertera temperaturen till Celsius och lagra den i temperatureCelsius fältet. Du kan också utöka källdata med fältet location från datauppsättningen för kontextualisering:

builtInTransformationSettings:
  map:
    - inputs:
      - temperature # - $1
      expression: "($1 - 32) * 5/9"
      output: temperatureCelsius
    - inputs:
      - $context(assetDataset).location  
      output: location

1. Under Transformera (valfritt) väljer du Ta bort.

2. Välj den datapunkt som du vill ta bort från utdataschemat.

   

3. Välj Använd.

Om du vill ta bort en datapunkt från utdataschemat kan du använda builtInTransformationSettings egenskapen i dataflödeskonfigurationen. Egenskapen map används för att ange de datapunkter som ska tas bort.

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

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

För driftupplevelse anger du schema- och serialiseringsformatet i information om dataflödesslutpunkten. Slutpunkterna som stöder serialiseringsformat är Microsoft Fabric OneLake, Azure Data Lake Storage Gen 2, Azure Data Explorer och lokal lagring. Om du till exempel vill serialisera data i Delta-format måste du ladda upp ett schema till schemaregistret och referera till dem i slutpunktskonfigurationen för dataflödet.

När du har laddat upp ett schema till schemaregistret kan du referera till det i dataflödeskonfigurationen.

{
  "builtInTransformationSettings": {
    "serializationFormat": "Delta",
    "schemaRef": "aio-sr://<SCHEMA_NAMESPACE>/<SCHEMA>:<VERSION>"
  }
}

När du har laddat upp ett schema till schemaregistret kan du referera till det i dataflödeskonfigurationen.

builtInTransformationSettings: {
  serializationFormat: 'Delta'
  schemaRef: 'aio-sr://<SCHEMA_NAMESPACE>/<SCHEMA>:<VERSION>'
}

När du har laddat upp ett schema till schemaregistret kan du referera till det i dataflödeskonfigurationen.

builtInTransformationSettings:
  serializationFormat: Delta
  schemaRef: 'aio-sr://<SCHEMA_NAMESPACE>/<SCHEMA>:<VERSION>'

1. Välj den dataflödesslutpunkt som ska användas som mål.

   

   Lagringsslutpunkter kräver ett schema för serialisering. Om du väljer en Microsoft Fabric OneLake-, Azure Data Lake Storage-, Azure Data Explorer- eller Local Storage-målslutpunkt måste du ange en schemareferens. Om du till exempel vill serialisera data till en Microsoft Fabric-slutpunkt i Delta-format måste du ladda upp ett schema till schemaregistret och referera till dem i dataflödets slutpunktskonfiguration.

   

2. Välj Fortsätt för att konfigurera målet.

3. Ange de inställningar som krävs för målet, inklusive det ämne eller den tabell som data ska skickas till. Mer information finns i Konfigurera datamål (ämne, container eller tabell).

{
  "destinationSettings": {
    "endpointRef": "<CUSTOM_ENDPOINT_NAME>",
    "dataDestination": "<TOPIC_OR_TABLE>" // See section on configuring data destination
  }
}

destinationSettings: {
  endpointRef: '<CUSTOM_ENDPOINT_NAME>'
  dataDestination: '<TOPIC_OR_TABLE>' // See section on configuring data destination
}

destinationSettings:
  endpointRef: <CUSTOM_ENDPOINT_NAME>
  dataDestination: <TOPIC_OR_TABLE> # See section on configuring data destination

När du använder driftupplevelsen tolkas datamålfältet automatiskt baserat på slutpunktstypen. Om dataflödesslutpunkten till exempel är en lagringsslutpunkt uppmanas du på sidan med målinformation att ange containernamnet. Om dataflödesslutpunkten är en MQTT-slutpunkt uppmanar målinformationssidan dig att ange ämnet och så vidare.

{
  "destinationSettings": {
    "endpointRef": "<CUSTOM_ENDPOINT_NAME>",
    "dataDestination": "<TOPIC_OR_TABLE>" // See section on configuring data destination
  }
}

Om du till exempel vill skicka data tillbaka till den lokala MQTT-koordinatorn via ett statiskt MQTT-ämne, använder du följande konfiguration:

{
  "destinationSettings": {
    "endpointRef": "default",
    "dataDestination": "example-topic"
  }
}

Om du har en anpassad slutpunkt för händelsehubben ser konfigurationen ut så här:

{
  "destinationSettings": {
    "endpointRef": "my-eh-endpoint",
    "dataDestination": "individual-event-hub"
  }
}

För MQTT-slutpunkter kan du också använda dynamiska ämnesvariabler. Om du till exempel vill dirigera meddelanden från factory/1/data till processed/factory/1:

{
  "destinationSettings": {
    "endpointRef": "default",
    "dataDestination": "processed/factory/${inputTopic.2}"
  }
}

Syntaxen är densamma för alla dataflödesslutpunkter:

destinationSettings: {
  endpointRef: "<CUSTOM_ENDPOINT_NAME>"
  dataDestination: '<TOPIC_OR_TABLE>'
}

Om du till exempel vill skicka data tillbaka till den lokala MQTT-koordinatorn via ett statiskt MQTT-ämne, använder du följande konfiguration:

destinationSettings: {
  endpointRef: 'default'
  dataDestination: 'example-topic'
}

Om du har en anpassad slutpunkt för händelsehubben ser konfigurationen ut så här:

destinationSettings: {
  endpointRef: 'my-eh-endpoint'
  dataDestination: 'individual-event-hub'
}

Ett annat exempel med en lagringsslutpunkt som mål:

destinationSettings: {
  endpointRef: 'my-adls-endpoint'
  dataDestination: 'my-container'
}

För MQTT-slutpunkter kan du också använda dynamiska ämnesvariabler:

destinationSettings: {
  endpointRef: 'default'
  dataDestination: 'processed/factory/${inputTopic.2}'
}

Syntaxen är densamma för alla dataflödesslutpunkter:

destinationSettings:
  endpointRef: <CUSTOM_ENDPOINT_NAME>
  dataDestination: <TOPIC_OR_TABLE>

Om du till exempel vill skicka data tillbaka till den lokala MQTT-koordinatorn via ett statiskt MQTT-ämne, använder du följande konfiguration:

destinationSettings:
  endpointRef: default
  dataDestination: example-topic

Om du har en anpassad slutpunkt för händelsehubben ser konfigurationen ut så här:

destinationSettings:
  endpointRef: my-eh-endpoint
  dataDestination: individual-event-hub

Ett annat exempel med en lagringsslutpunkt som mål:

destinationSettings:
  endpointRef: my-adls-endpoint
  dataDestination: my-container

För MQTT-slutpunkter kan du också använda dynamiska ämnesvariabler:

destinationSettings:
  endpointRef: default
  dataDestination: processed/factory/${inputTopic.2}

Använd kommandot az iot ops dataflow apply för att skapa eller ändra ett dataflöde.

az iot ops dataflow apply --resource-group <ResourceGroupName> --instance <AioInstanceName> --profile <DataflowProfileName> --name <DataflowName> --config-file <ConfigFilePathAndName>

Parametern --config-file är sökvägen och filnamnet för en JSON-konfigurationsfil som innehåller resursegenskaperna.

I det här exemplet antar du att en konfigurationsfil med namnet data-flow.json med följande innehåll lagras i användarens hemkatalog:

{
    "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": "fToC($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"
      }
    ]
}

Här är ett exempelkommando för att skapa eller uppdatera ett dataflöde med hjälp av standardprofilen för dataflöde:

az iot ops dataflow apply --resource-group myResourceGroup --instance myAioInstance --profile default --name data-flow --config-file ~/data-flow.json

Här är ett annat exempel som använder dynamisk ämnesöversättning för att dirigera meddelanden från olika termostater till enhetsspecifika ämnen:

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

Den här konfigurationen bearbetar meddelanden från thermostats/device1/sensor/temperature och skickar dem till 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@2024-11-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@2024-11-01' existing = {
  parent: aioInstance
  name: 'default'
}

// Pointer to the default data flow profile
resource defaultDataflowProfile 'Microsoft.IoTOperations/instances/dataflowProfiles@2024-11-01' existing = {
  parent: aioInstance
  name: 'default'
}

resource dataflow 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflows@2024-11-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'
        }
      }
    ]
  }
}

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

Välj det dataflöde som du vill exportera och välj Exportera i verktygsfältet.

Använd kommandot az iot ops dataflow show för att exportera ett dataflöde.

az iot ops dataflow show --resource-group <ResourceGroupName> --instance <AioInstanceName> --name <DataflowName> --profile <DataflowProfileName> --output json > my-dataflow.json

Här är ett exempelkommando för att exportera ett dataflöde med namnet data-flow till en JSON-fil med namnet data-flow.json:

az iot ops dataflow show --resource-group myResourceGroup --instance myAioInstance --profile default --name data-flow --output json > data-flow.json

Bicep är infrastruktur som kod och ingen export krävs. Använd Bicep-filen för att skapa ett dataflöde för att snabbt konfigurera dataflöden.

kubectl get dataflow my-dataflow -o yaml > my-dataflow.yaml