Konfigurera dataflöden i Azure IoT Operations

Viktigt!

Förhandsversion av Azure IoT Operations – aktiverad av Azure Arc är för närvarande i förhandsversion. Du bör inte använda den här förhandsgranskningsprogramvaran i produktionsmiljöer.

Du måste distribuera en ny Azure IoT Operations-installation när en allmänt tillgänglig version görs tillgänglig. Du kommer inte att kunna uppgradera en förhandsgranskningsinstallation.

Juridiska villkor för Azure-funktioner i betaversion, förhandsversion eller som av någon annan anledning inte har gjorts allmänt tillgängliga ännu finns i kompletterande användningsvillkor för Microsoft Azure-förhandsversioner.

Ett dataflöde är den sökväg som data tar från källan till målet med valfria transformeringar. Du kan konfigurera dataflödet genom att skapa en anpassad dataflödesresurs eller med hjälp av Azure IoT Operations Studio-portalen. Ett dataflöde består av tre delar: källan, omvandlingen och målet.

För att definiera källan och målet måste du konfigurera dataflödesslutpunkterna. Omvandlingen är valfri och kan omfatta åtgärder som att berika data, filtrera data och mappa data till ett annat fält.

Viktigt!

Varje dataflöde måste ha standardslutpunkten för Azure IoT Operations lokal MQTT-koordinator som antingen källa eller mål.

Du kan använda driftupplevelsen i Azure IoT Operations för att skapa ett dataflöde. Driftupplevelsen tillhandahåller ett visuellt gränssnitt för att konfigurera dataflödet. Du kan också använda Bicep för att skapa ett dataflöde med en Bicep-mallfil eller använda Kubernetes för att skapa ett dataflöde med hjälp av en YAML-fil.

Fortsätt läsa för att lära dig hur du konfigurerar källan, omvandlingen och målet.

Förutsättningar

Du kan distribuera dataflöden så snart du har en instans av Azure IoT Operations Preview med hjälp av standardprofilen för dataflöde och slutpunkten. Du kanske dock vill konfigurera dataflödesprofiler och slutpunkter för att anpassa dataflödet.

Dataflödesprofil

Dataflödesprofilen anger antalet instanser för de dataflöden under den som ska användas. Om du inte behöver flera grupper med dataflöden med olika skalningsinställningar kan du använda standardprofilen för dataflöde. Information om hur du konfigurerar en dataflödesprofil finns i Konfigurera dataflödesprofiler.

Dataflödesslutpunkter

Dataflödesslutpunkter krävs för att konfigurera källan och målet för dataflödet. För att komma igång snabbt kan du använda standardslutpunkten för dataflöde för den lokala MQTT-koordinatorn. Du kan också skapa andra typer av dataflödesslutpunkter som Kafka, Event Hubs eller Azure Data Lake Storage. Information om hur du konfigurerar varje typ av dataflödesslutpunkt finns i Konfigurera dataflödesslutpunkter.

Kom igång

När du har förutsättningar kan du börja skapa ett dataflöde.

Portal

Om du vill skapa ett dataflöde i driftmiljön väljer du Dataflöde>Skapa dataflöde. Sedan ser du sidan där du kan konfigurera källan, omvandlingen och målet för dataflödet.

Bicep

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-09-15-preview' existing = {
  name: aioInstanceName
}

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

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

// Pointer to the default dataflow profile
resource defaultDataflowProfile 'Microsoft.IoTOperations/instances/dataflowProfiles@2024-09-15-preview' existing = {
  parent: aioInstance
  name: 'default'
}

resource dataflow 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflows@2024-09-15-preview' = {
  // Reference to the parent dataflow 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
        }
      }
    ]
  }
}

Kubernetes

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/v1beta1
kind: Dataflow
metadata:
  name: <DATAFLOW_NAME>
  namespace: azure-iot-operations
spec:
  # Reference to the default dataflow 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

Läs följande avsnitt om du vill lära dig hur du konfigurerar åtgärdstyperna för dataflödet.

Källa

Om du vill konfigurera en källa för dataflödet anger du slutpunktsreferensen och en lista över datakällor för slutpunkten.

Använda tillgången som källa

Portal

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.

Bicep

Det är bara tillgängligt att konfigurera en tillgång som källa i driftmiljön.

Kubernetes

Det är bara tillgängligt att konfigurera en tillgång som källa i driftmiljön.

När du använder en tillgång som källa används tillgångsdefinitionen för att härleda schemat för dataflödet. Tillgångsdefinitionen innehåller schemat för tillgångens datapunkter. Mer information finns i Hantera tillgångskonfigurationer via fjärranslutning.

När data från tillgången har konfigurerats nådde de dataflödet via den lokala MQTT-koordinatorn. När du använder en tillgång som källa använder dataflödet därför den lokala standardslutpunkten för MQTT-koordinatorn som källa i verkligheten.

Använda MQTT-standardslutpunkten som källa

Portal

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

   

2. Ange följande inställningar för MQTT-källan:

   Inställning	beskrivning
   MQTT-ämne	MQTT-ämnesfiltret som du vill prenumerera på för inkommande meddelanden. Se Avsnittet Konfigurera MQTT eller Kafka.
   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.

Bicep

MQTT-slutpunkten konfigureras i Bicep-mallfilen. Följande slutpunkt är till exempel en källa för dataflödet.

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

dataSources Här kan du ange flera MQTT- eller Kafka-ämnen utan att behöva ändra slutpunktskonfigurationen. Den här flexibiliteten innebär att samma slutpunkt kan återanvändas i flera dataflöden, även om ämnena varierar. Mer information finns i Konfigurera datakällor.

Kubernetes

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

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

Eftersom dataSources du kan ange MQTT- eller Kafka-ämnen utan att ändra slutpunktskonfigurationen kan du återanvända slutpunkten för flera dataflöden även om ämnena är olika. Mer information finns i Konfigurera datakällor.

Om standardslutpunkten inte används som källa måste den användas som mål. Mer information finns i Dataflöden måste använda den lokala MQTT-koordinatorslutpunkten.

Använda anpassad MQTT- eller Kafka-dataflödesslutpunkt som källa

Om du har skapat en anpassad MQTT- eller Kafka-dataflödesslutpunkt (till exempel för användning med Event Grid eller Event Hubs) kan du använda den som källa för dataflödet. Kom ihåg att slutpunkter av lagringstyp, till exempel Data Lake eller Fabric OneLake, inte kan användas som källa.

För att konfigurera använder du Kubernetes YAML eller Bicep. Ersätt platshållarvärden med ditt anpassade slutpunktsnamn och ämnen.

Portal

Användning av en anpassad MQTT- eller Kafka-slutpunkt som källa stöds för närvarande inte i driftupplevelsen.

Bicep

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

Kubernetes

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

Konfigurera datakällor (MQTT- eller Kafka-ämnen)

Du kan ange flera MQTT- eller Kafka-ämnen i en källa utan att behöva ändra dataflödesslutpunktskonfigurationen. Den här flexibiliteten innebär att samma slutpunkt kan återanvändas i flera dataflöden, även om ämnena varierar. Mer information finns i Återanvända dataflödesslutpunkter.

MQTT-ämnen

När källan är en MQTT-slutpunkt (Event Grid ingår) kan du använda MQTT-ämnesfiltret för att prenumerera på inkommande meddelanden. Ämnesfiltret kan innehålla jokertecken för att prenumerera på flera ämnen. Till exempel thermostats/+/telemetry/temperature/# prenumererar på alla temperaturtelemetrimeddelanden från termostater. Så här konfigurerar du MQTT-ämnesfilter:

Portal

I informationen om dataflödets driftupplevelse väljer du MQTT och använder sedan fältet MQTT-ämne för att ange MQTT-ämnesfiltret för att prenumerera på inkommande meddelanden.

Kommentar

Endast ett MQTT-ämnesfilter kan anges i driftupplevelsen. Om du vill använda flera MQTT-ämnesfilter använder du Bicep eller Kubernetes.

Bicep

sourceSettings: {
  endpointRef: '<MQTT_ENDPOINT_NAME>'
  dataSources: [
    '<MQTT_TOPIC_FILTER_1>'
    '<MQTT_TOPIC_FILTER_2>'
    // Add more MQTT topic filters as needed
  ]
}

Exempel med flera MQTT-ämnesfilter med jokertecken:

sourceSettings: {
  endpointRef: 'default'
  dataSources: [
    'thermostats/+/telemetry/temperature/#'
    'humidifiers/+/telemetry/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 telemetrimeddelanden under alla underämnen i ämnena temperature och humidity .

Kubernetes

sourceSettings:
  endpointRef: <MQTT_ENDPOINT_NAME>
  dataSources:
    - <MQTT_TOPIC_FILTER_1>
    - <MQTT_TOPIC_FILTER_2>
    # Add more MQTT topic filters as needed

Exempel med flera MQTT-ämnesfilter med jokertecken:

sourceSettings:
  endpointRef: default
  dataSources:
    - thermostats/+/telemetry/temperature/#
    - humidifiers/+/telemetry/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 telemetrimeddelanden under alla underämnen i ämnena temperature och humidity .

Delade prenumerationer

Om du vill använda delade prenumerationer med MQTT-källor kan du ange det delade prenumerationsavsnittet i form av $shared/<GROUP_NAME>/<TOPIC_FILTER>.

Portal

I operations experience dataflow Source details (Dataflödeskälla ) väljer du MQTT och använder fältet MQTT-ämne för att ange den delade prenumerationsgruppen och ämnet.

Bicep

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

Kubernetes

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

Om instansantalet i dataflödesprofilen är större än 1 aktiveras den delade prenumerationen automatiskt för alla dataflöden som använder MQTT-källa. I det här fallet läggs prefixet $shared till och namnet på den delade prenumerationsgruppen genereras automatiskt. Om du till exempel har en dataflödesprofil med instansantalet 3 och ditt dataflöde använder en MQTT-slutpunkt som källa konfigurerad med ämnen topic1 och topic2konverteras de automatiskt till delade prenumerationer som $shared/<GENERATED_GROUP_NAME>/topic1 och $shared/<GENERATED_GROUP_NAME>/topic2. Om du vill använda ett annat grupp-ID för delad prenumeration kan du åsidosätta det i ämnet, till exempel $shared/mygroup/topic1.

Viktigt!

Dataflöden som kräver delad prenumeration när antalet instanser är större än 1 är viktigt när du använder Event Grid MQTT-koordinator som källa eftersom det inte stöder delade prenumerationer. Om du vill undvika att meddelanden saknas anger du antalet instanser av dataflödesprofilen till 1 när du använder Event Grid MQTT-koordinatorn som källa. Det är då dataflödet är prenumeranten och tar emot meddelanden från molnet.

Kafka-ämnen

När källan är en Kafka-slutpunkt (inklusive Event Hubs) anger du de enskilda kafka-ämnen som ska prenumerera på inkommande meddelanden. Jokertecken stöds inte, så du måste ange varje ämne statiskt.

Kommentar

När du använder Event Hubs via Kafka-slutpunkten är varje enskild händelsehubb i namnområdet Kafka-ämnet. Om du till exempel har ett Event Hubs-namnområde med två händelsehubbar thermostats och humidifierskan du ange varje händelsehubb som ett Kafka-ämne.

Så här konfigurerar du Kafka-ämnena:

Portal

Användning av en Kafka-slutpunkt som källa stöds för närvarande inte i driftupplevelsen.

Bicep

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

Kubernetes

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

Ange schema för att deserialisera data

Om källdata har valfria fält eller fält med olika typer anger du ett deserialiseringsschema för att säkerställa konsekvens. Data kan till exempel ha fält som inte finns i alla meddelanden. Utan schemat kan transformeringen inte hantera dessa fält eftersom de skulle ha tomma värden. Med schemat kan du ange standardvärden eller ignorera fälten.

Att ange schemat är bara relevant när du använder MQTT- eller Kafka-källan. Om källan är en tillgång härleds schemat automatiskt från tillgångsdefinitionen.

Så här konfigurerar du schemat som används för att deserialisera inkommande meddelanden från en källa:

Portal

I information om dataflödets dataflöde väljer du MQTT och använder fältet Meddelandeschema för att ange schemat. Du kan använda knappen Ladda upp för att ladda upp en schemafil först. Mer information finns i Förstå meddelandescheman.

Bicep

När du har använt 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>'
}

Kubernetes

När du har använt 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>'

Transformering

Transformeringsåtgärden är den plats där du kan transformera data från källan innan du skickar dem till målet. Transformeringar är valfria. Om du inte behöver göra ändringar i data ska du inte inkludera transformeringsåtgärden i dataflödeskonfigurationen. Flera transformeringar länkas samman i steg oavsett i vilken ordning de anges i konfigurationen. Stegens ordning är alltid:

1. Berika, byt namn på eller lägg till en ny egenskap: Lägg till ytterligare data i källdata med en datauppsättning och ett villkor som ska matchas.
2. Filter: Filtrera data baserat på ett villkor.
3. Mappa eller beräkna: Flytta data från ett fält till ett annat med en valfri konvertering.

Portal

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

Bicep

builtInTransformationSettings: {
  datasets: [
    // See section on enriching data
  ]
  filter: [
    // See section on filtering data
  ]
  map: [
    // See section on mapping data
  ]
}

Kubernetes

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

Berika: Lägga till referensdata

Om du vill utöka data kan du använda referensdatauppsättningen i Azure IoT Operations distributed state store (DSS). Datauppsättningen används för att lägga till extra data i källdata baserat på ett villkor. Villkoret anges som ett fält i källdata som matchar ett fält i datamängden.

Du kan läsa in exempeldata i DSS med hjälp av DSS-uppsättningens verktygsexempel. Nyckelnamn i det distribuerade tillståndslagret motsvarar en datauppsättning i dataflödeskonfigurationen.

Portal

I driftupplevelsen stöds för närvarande Enrich-fasen med hjälp av transformeringarna Byt namn och Ny.

1. I driftupplevelsen väljer du ett dataflöde och sedan Lägg till transformering (valfritt).

2. Välj Byt namn eller Nya egenskapstransformeringar och välj sedan Lägg till.

   

Bicep

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

Kubernetes

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

Om datamängden har en post med fältet asset , ungefär som:

{
  "asset": "thermostat1",
  "location": "room1",
  "manufacturer": "Contoso"
}

Data från källan med deviceId fältmatchning thermostat1 har fälten location och manufacturer tillgängliga i filter- och kartfaser.

Mer information om villkorssyntax finns i Berika data med hjälp av dataflöden och Konvertera data med hjälp av dataflöden.

Filter: Filtrera data baserat på ett villkor

Om du vill filtrera data på ett villkor kan du använda filter fasen. Villkoret anges som ett fält i källdata som matchar ett värde.

Portal

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

2. Välj de datapunkter som ska ingå i datauppsättningen.

3. Lägg till ett filtervillkor och en beskrivning.

   

4. Välj Använd.

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.

Bicep

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.

Kubernetes

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.

Karta: Flytta data från ett fält till ett annat

Om du vill mappa data till ett annat fält med valfri konvertering kan du använda åtgärden map . Konverteringen anges som en formel som använder fälten i källdata.

Portal

I driftmiljön stöds mappning för närvarande med hjälp av Beräkningstransformeringar .

1. Under Transformera (valfritt) väljer du Beräkningstillägg>.

2. Ange de obligatoriska fälten och uttrycken.

   

3. Välj Använd.

Bicep

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

Kubernetes

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

Mer information finns i Mappa data med hjälp av dataflöden och Konvertera data med hjälp av dataflöden.

Serialisera data enligt ett schema

Om du vill serialisera data innan du skickar dem till målet måste du ange ett schema- och serialiseringsformat. Annars serialiseras data i JSON med de typer som härleds. Lagringsslutpunkter som Microsoft Fabric eller Azure Data Lake kräver ett schema för att säkerställa datakonsekvens. Serialiseringsformat som stöds är Parquet och Delta.

Portal

För närvarande stöds inte att ange utdataschemat och serialiseringen i driftmiljön.

Bicep

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

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

Kubernetes

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

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

Mer information om schemaregistret finns i Förstå meddelandescheman.

Mål

Om du vill konfigurera ett mål för dataflödet anger du slutpunktsreferensen och datamålet. Du kan ange en lista över datamål för slutpunkten.

Om du vill skicka data till ett annat mål än den lokala MQTT-koordinatorn skapar du en dataflödesslutpunkt. Mer information finns i Konfigurera dataflödesslutpunkter. Om målet inte är den lokala MQTT-koordinatorn måste det användas som källa. Mer information finns i Dataflöden måste använda den lokala MQTT-koordinatorslutpunkten.

Viktigt!

Lagringsslutpunkter kräver en schemareferens. Om du har skapat slutpunkter för lagringsmål för Microsoft Fabric OneLake, ADLS Gen 2, Azure Data Explorer och Lokal lagring måste du ange schemareferens.

Portal

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

   

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).

Bicep

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

Kubernetes

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

Konfigurera datamål (ämne, container eller tabell)

Precis som datakällor är datamål ett begrepp som används för att hålla dataflödesslutpunkterna återanvändbara över flera dataflöden. I princip representerar den underkatalogen i konfigurationen av dataflödesslutpunkten. Om dataflödesslutpunkten till exempel är en lagringsslutpunkt är datamålet tabellen i lagringskontot. Om dataflödesslutpunkten är en Kafka-slutpunkt är datamålet Kafka-ämnet.

Slutpunktstyp	Betydelse för datamål	beskrivning
MQTT (eller Event Grid)	Område	MQTT-ämnet där data skickas. Endast statiska ämnen stöds, inga jokertecken.
Kafka (eller Event Hubs)	Område	Kafka-ämnet där data skickas. Endast statiska ämnen stöds, inga jokertecken. Om slutpunkten är ett Event Hubs-namnområde är datamålet den enskilda händelsehubben i namnområdet.
Azure Data Lake Storage	Container	Containern i lagringskontot. Inte tabellen.
Microsoft Fabric OneLake	Tabell eller mapp	Motsvarar den konfigurerade sökvägstypen för slutpunkten.
Öppna Azure-datautforskaren	Register	Tabellen i Azure Data Explorer-databasen.
Lokal lagring	Mapp	Mapp- eller katalognamnet i den lokala lagringsbeständiga volymmonteringen. När du använder Azure Container Storage som aktiveras av Azure Arc Cloud Ingest Edge-volymer måste detta matcha parametern spec.path för den delvolym som du skapade.

Så här konfigurerar du datamålet:

Portal

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.

Bicep

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

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

Om du till exempel vill skicka tillbaka data till den lokala MQTT-koordinatorn 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'
}

Kubernetes

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

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

Om du till exempel vill skicka tillbaka data till den lokala MQTT-koordinatorn 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

Exempel

Följande exempel är en dataflödeskonfiguration som använder MQTT-slutpunkten för källan och målet. Källan filtrerar data från MQTT-ämnet azure-iot-operations/data/thermostat. Omvandlingen konverterar temperaturen till Fahrenheit och filtrerar data där temperaturen multiplicerat med fuktigheten är mindre än 100000. Målet skickar data till MQTT-ämnet factory.

Portal

Se flikarna Bicep eller Kubernetes för konfigurationsexemplet.

Bicep

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

resource aioInstance 'Microsoft.IoTOperations/instances@2024-09-15-preview' existing = {
  name: aioInstanceName
}

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

// Pointer to the default dataflow endpoint
resource defaultDataflowEndpoint 'Microsoft.IoTOperations/instances/dataflowEndpoints@2024-09-15-preview' existing = {
  parent: aioInstance
  name: 'default'
}

// Pointer to the default dataflow profile
resource defaultDataflowProfile 'Microsoft.IoTOperations/instances/dataflowProfiles@2024-09-15-preview' existing = {
  parent: aioInstance
  name: 'default'
}

resource dataflow 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflows@2024-09-15-preview' = {
  // Reference to the parent dataflow 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'
        }
      }
    ]
  }
}

Kubernetes

apiVersion: connectivity.iotoperations.azure.com/v1beta1
kind: Dataflow
metadata:
  name: my-dataflow
  namespace: azure-iot-operations
spec:
  # Reference to the default dataflow 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

Mer information om dataflödeskonfigurationer finns i Azure REST API – Dataflöde och snabbstarten Bicep.

Kontrollera att ett dataflöde fungerar

Följ självstudie: Dubbelriktad MQTT-brygga till Azure Event Grid för att kontrollera att dataflödet fungerar.

Exportera dataflödeskonfiguration

Om du vill exportera dataflödeskonfigurationen kan du använda driftupplevelsen eller genom att exportera den anpassade dataflödesresursen.

Portal

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

Bicep

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

Kubernetes

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

Korrekt dataflödeskonfiguration

Kontrollera följande för att säkerställa att dataflödet fungerar som förväntat:

• Standardslutpunkten för MQTT-dataflöde måste användas som källa eller mål.
• Dataflödesprofilen finns och refereras till i dataflödeskonfigurationen.
• Källan är antingen en MQTT-slutpunkt, Kafka-slutpunkt eller en tillgång. Slutpunkter av lagringstyp kan inte användas som källa.
• När du använder Event Grid som källa anges instansantalet för dataflödesprofilen till 1 eftersom Event Grid MQTT-asynkron meddelandekö inte stöder delade prenumerationer.
• När du använder Event Hubs som källa är varje händelsehubb i namnområdet ett separat Kafka-ämne och måste anges som datakälla.
• Transformering, om den används, konfigureras med rätt syntax, inklusive korrekt undflyende av specialtecken.
• När du använder slutpunkter av lagringstyp som mål anges ett schema.

Nästa steg

• Mappa data med hjälp av dataflöden
• Konvertera data med hjälp av dataflöden
• Berika data med hjälp av dataflöden
• Förstå meddelandescheman
• Hantera dataflödesprofiler