Gegevensstromen configureren in Azure IoT-bewerkingen

Artikel
12/12/2024

Belangrijk

Deze pagina bevat instructies voor het beheren van Azure IoT Operations-onderdelen met behulp van Kubernetes-implementatiemanifesten, die in preview zijn. Deze functie is voorzien van verschillende beperkingen en mag niet worden gebruikt voor productieworkloads.

Raadpleeg de Aanvullende voorwaarden voor Microsoft Azure-previews voor juridische voorwaarden die van toepassing zijn op Azure-functies die in bèta of preview zijn of die anders nog niet algemeen beschikbaar zijn.

Een gegevensstroom is het pad dat gegevens van de bron naar de bestemming halen met optionele transformaties. U kunt de gegevensstroom configureren door een aangepaste gegevensstroomresource te maken of met behulp van de Azure IoT Operations Studio-portal. Een gegevensstroom bestaat uit drie delen: de bron, de transformatie en de bestemming.

Diagram van een gegevensstroom met stroom van bron naar transformatie en bestemming.

Als u de bron en het doel wilt definiëren, moet u de eindpunten voor de gegevensstroom configureren. De transformatie is optioneel en kan bewerkingen bevatten, zoals het verrijken van de gegevens, het filteren van de gegevens en het toewijzen van de gegevens aan een ander veld.

Belangrijk

Elke gegevensstroom moet het standaardeindpunt van de azure IoT Operations-MQTT-broker hebben als de bron of het doel.

U kunt de bewerkingservaring in Azure IoT-bewerkingen gebruiken om een gegevensstroom te maken. De bewerkingservaring biedt een visuele interface voor het configureren van de gegevensstroom. U kunt bicep ook gebruiken om een gegevensstroom te maken met behulp van een Bicep-sjabloonbestand of Kubernetes gebruiken om een gegevensstroom te maken met behulp van een YAML-bestand.

Lees verder voor meer informatie over het configureren van de bron, transformatie en bestemming.

Vereisten

U kunt gegevensstromen implementeren zodra u een exemplaar van Azure IoT Operations hebt met behulp van het standaardgegevensstroomprofiel en -eindpunt. Het is echter mogelijk dat u gegevensstroomprofielen en -eindpunten wilt configureren om de gegevensstroom aan te passen.

Gegevensstroomprofiel

Als u geen andere schaalinstellingen voor uw gegevensstromen nodig hebt, gebruikt u het standaardgegevensstroomprofiel van Azure IoT Operations. Zie Gegevensstroomprofielen configureren voor meer informatie over het configureren van een gegevensstroomprofiel.

Gegevensstroomeindpunten

Eindpunten voor gegevensstromen zijn vereist om de bron en het doel voor de gegevensstroom te configureren. Als u snel aan de slag wilt gaan, kunt u het standaardgegevensstroomeindpunt voor de lokale MQTT-broker gebruiken. U kunt ook andere typen gegevensstroomeindpunten maken, zoals Kafka, Event Hubs of Azure Data Lake Storage. Zie Eindpunten voor gegevensstromen configureren voor meer informatie over het configureren van elk type gegevensstroomeindpunt.

Aan de slag

Zodra u aan de vereisten voldoet, kunt u beginnen met het maken van een gegevensstroom.

Als u een gegevensstroom in de bewerkingservaring wilt maken, selecteert u Gegevensstroom> maken. Vervolgens ziet u de pagina waar u de bron, transformatie en bestemming voor de gegevensstroom kunt configureren.

Schermopname van bewerkingen om een gegevensstroom te maken.

Maak een Bicep-bestand .bicep om een gegevensstroom te maken. In dit voorbeeld ziet u de structuur van de gegevensstroom met de bron-, transformatie- en doelconfiguraties.

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

Maak een Kubernetes-manifestbestand .yaml om een gegevensstroom te maken. In dit voorbeeld ziet u de structuur van de gegevensstroom met de bron-, transformatie- en doelconfiguraties.

apiVersion: connectivity.iotoperations.azure.com/v1
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

Bekijk de volgende secties voor meer informatie over het configureren van de bewerkingstypen van de gegevensstroom.

Bron

Als u een bron voor de gegevensstroom wilt configureren, geeft u de eindpuntverwijzing en een lijst met gegevensbronnen voor het eindpunt op. Kies een van de volgende opties als bron voor de gegevensstroom.

Als het standaardeindpunt niet als bron wordt gebruikt, moet het als doel worden gebruikt. Zie Gegevensstromen moeten lokaal MQTT-brokereindpunt gebruiken voor meer informatie.

Optie 1: Standaard MQTT-eindpunt gebruiken als bron

Selecteer onder Brondetails de optie MQTT.

Voer de volgende instellingen in voor de MQTT-bron:

Instelling	Beschrijving
MQTT-onderwerp	Het MQTT-onderwerpfilter waarop u zich wilt abonneren voor binnenkomende berichten. Zie MQTT- of Kafka-onderwerpen configureren.
Berichtschema	Het schema dat moet worden gebruikt om de binnenkomende berichten te deserialiseren. Zie Schema opgeven om gegevens te deserialiseren.

Selecteer Toepassen.

Het MQTT-eindpunt is geconfigureerd in het Bicep-sjabloonbestand. Het volgende eindpunt is bijvoorbeeld een bron voor de gegevensstroom.

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

dataSources Hier kunt u meerdere MQTT- of Kafka-onderwerpen opgeven zonder dat u de eindpuntconfiguratie hoeft te wijzigen. Deze flexibiliteit betekent dat hetzelfde eindpunt opnieuw kan worden gebruikt voor meerdere gegevensstromen, zelfs als de onderwerpen verschillen. Zie Gegevensbronnen configureren voor meer informatie.

Als u bijvoorbeeld een bron wilt configureren met behulp van een MQTT-eindpunt en twee MQTT-onderwerpfilters, gebruikt u de volgende configuratie:

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

Omdat dataSources u MQTT- of Kafka-onderwerpen kunt opgeven zonder de eindpuntconfiguratie te wijzigen, kunt u het eindpunt opnieuw gebruiken voor meerdere gegevensstromen, zelfs als de onderwerpen verschillend zijn. Zie Gegevensbronnen configureren voor meer informatie.

Optie 2: Asset als bron gebruiken

U kunt een asset als bron voor de gegevensstroom gebruiken. Het gebruik van een asset als bron is alleen beschikbaar in de bewerkingservaring.

Selecteer Asset onder Brondetails.
Selecteer de asset die u wilt gebruiken als het broneindpunt.
Selecteer Doorgaan.

Er wordt een lijst met gegevenspunten voor de geselecteerde asset weergegeven.
Selecteer Toepassen om de asset als broneindpunt te gebruiken.

Wanneer u een asset als bron gebruikt, wordt de assetdefinitie gebruikt om het schema voor de gegevensstroom af te leiden. De assetdefinitie bevat het schema voor de gegevenspunten van de asset. Zie Assetconfiguraties op afstand beheren voor meer informatie.

Zodra de gegevens van de asset zijn geconfigureerd, hebben ze de gegevensstroom bereikt via de lokale MQTT-broker. Wanneer u dus een asset als bron gebruikt, gebruikt de gegevensstroom het standaardeindpunt van de lokale MQTT-broker als de bron in werkelijkheid.

Optie 3: Aangepast MQTT- of Kafka-gegevensstroomeindpunt als bron gebruiken

Als u een aangepast MQTT- of Kafka-gegevensstroomeindpunt hebt gemaakt (bijvoorbeeld voor gebruik met Event Grid of Event Hubs), kunt u dit gebruiken als bron voor de gegevensstroom. Houd er rekening mee dat eindpunten van het opslagtype, zoals Data Lake of Fabric OneLake, niet als bron kunnen worden gebruikt.

Gebruik Kubernetes YAML of Bicep om te configureren. Vervang tijdelijke aanduidingen door de naam en onderwerpen van uw aangepaste eindpunt.

Het gebruik van een aangepast MQTT- of Kafka-eindpunt als bron wordt momenteel niet ondersteund in de bewerkingservaring.

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

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

Gegevensbronnen (MQTT- of Kafka-onderwerpen) configureren

U kunt meerdere MQTT- of Kafka-onderwerpen in een bron opgeven zonder dat u de configuratie van het gegevensstroomeindpunt hoeft te wijzigen. Deze flexibiliteit betekent dat hetzelfde eindpunt opnieuw kan worden gebruikt voor meerdere gegevensstromen, zelfs als de onderwerpen verschillen. Zie Eindpunten voor gegevensstromen opnieuw gebruiken voor meer informatie.

MQTT-onderwerpen

Wanneer de bron een MQTT-eindpunt (Opgenomen Event Grid) is, kunt u het MQTT-onderwerpfilter gebruiken om u te abonneren op binnenkomende berichten. Het onderwerpfilter kan jokertekens bevatten om u te abonneren op meerdere onderwerpen. Abonneert zich bijvoorbeeld thermostats/+/telemetry/temperature/# op alle telemetrieberichten van temperatuur van thermostaten. De MQTT-onderwerpfilters configureren:

Selecteer MQTT in de gegevensstroombrondetails van de bewerkingservaring en gebruik vervolgens het onderwerpveld MQTT om het MQTT-onderwerpfilter op te geven waarop u zich wilt abonneren voor binnenkomende berichten.

Notitie

Er kan slechts één MQTT-onderwerpfilter worden opgegeven in de bewerkingservaring. Als u meerdere MQTT-onderwerpfilters wilt gebruiken, gebruikt u Bicep of Kubernetes.

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

Voorbeeld met meerdere MQTT-onderwerpfilters met jokertekens:

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

Hier wordt het jokerteken + gebruikt om alle apparaten onder de thermostats en humidifiers onderwerpen te selecteren. Het # jokerteken wordt gebruikt om alle telemetrieberichten onder alle subonderwerpen van de temperature en humidity onderwerpen te selecteren.

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

Voorbeeld met meerdere MQTT-onderwerpfilters met jokertekens:

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

Gedeelde abonnementen

Als u gedeelde abonnementen wilt gebruiken met MQTT-bronnen, kunt u het onderwerp voor gedeelde abonnementen opgeven in de vorm van $shared/<GROUP_NAME>/<TOPIC_FILTER>.

Selecteer MQTT in operations experience dataflow Source details en gebruik het MQTT-onderwerpveld om de gedeelde abonnementsgroep en het onderwerp op te geven.

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

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

Als het aantal exemplaren in het gegevensstroomprofiel groter is dan één, wordt het gedeelde abonnement automatisch ingeschakeld voor alle gegevensstromen die gebruikmaken van de MQTT-bron. In dit geval wordt het $shared voorvoegsel toegevoegd en wordt automatisch de naam van de gedeelde abonnementsgroep gegenereerd. Als u bijvoorbeeld een gegevensstroomprofiel hebt met het aantal exemplaren van 3 en uw gegevensstroom een MQTT-eindpunt gebruikt als bron die is geconfigureerd met onderwerpen topic1 en topic2, worden ze automatisch geconverteerd naar gedeelde abonnementen als $shared/<GENERATED_GROUP_NAME>/topic1 en $shared/<GENERATED_GROUP_NAME>/topic2.

U kunt expliciet een onderwerp maken met de naam $shared/mygroup/topic in uw configuratie. Het toevoegen van het $shared onderwerp wordt echter niet expliciet aanbevolen, omdat het $shared voorvoegsel automatisch wordt toegevoegd wanneer dat nodig is. Gegevensstromen kunnen optimalisaties maken met de groepsnaam als deze niet is ingesteld. Is bijvoorbeeld $share niet ingesteld en gegevensstromen hoeven alleen te worden uitgevoerd via de onderwerpnaam.

Belangrijk

Gegevensstromen waarvoor een gedeeld abonnement is vereist wanneer het aantal exemplaren groter is dan één, is belangrijk bij het gebruik van Event Grid MQTT Broker als bron, omdat het geen ondersteuning biedt voor gedeelde abonnementen. Als u ontbrekende berichten wilt voorkomen, stelt u het aantal exemplaren van het gegevensstroomprofiel in op één wanneer u Event Grid MQTT-broker als bron gebruikt. Dat is wanneer de gegevensstroom de abonnee is en berichten ontvangt van de cloud.

Kafka-onderwerpen

Wanneer de bron een Kafka-eindpunt (opgenomen Event Hubs) is, geeft u de afzonderlijke Kafka-onderwerpen op waarop u zich wilt abonneren voor binnenkomende berichten. Jokertekens worden niet ondersteund, dus u moet elk onderwerp statisch opgeven.

Notitie

Wanneer u Event Hubs via het Kafka-eindpunt gebruikt, is elke afzonderlijke Event Hub binnen de naamruimte het Kafka-onderwerp. Als u bijvoorbeeld een Event Hubs-naamruimte met twee Event Hubs-hubs thermostats hebt en humidifiersu kunt elke Event Hub opgeven als een Kafka-onderwerp.

De Kafka-onderwerpen configureren:

Het gebruik van een Kafka-eindpunt als bron wordt momenteel niet ondersteund in de bewerkingservaring.

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

Bronschema opgeven

Wanneer u MQTT of Kafka als bron gebruikt, kunt u een schema opgeven om de lijst met gegevenspunten weer te geven in de operations experience-portal. Houd er rekening mee dat het gebruik van een schema voor het deserialiseren en valideren van binnenkomende berichten momenteel niet wordt ondersteund.

Als de bron een asset is, wordt het schema automatisch afgeleid van de assetdefinitie.

Tip

Als u het schema wilt genereren op basis van een voorbeeldgegevensbestand, gebruikt u de Helper voor Schema Gen.

Het schema configureren dat wordt gebruikt om de binnenkomende berichten van een bron te deserialiseren:

Selecteer in bewerkingen gegevensstroombrondetails MQTT en gebruik het veld Berichtschema om het schema op te geven. U kunt de knop Uploaden gebruiken om eerst een schemabestand te uploaden. Zie Berichtschema's begrijpen voor meer informatie.

Zodra u het schemaregister hebt gebruikt om het schema op te slaan, kunt u ernaar verwijzen in de configuratie van de gegevensstroom.

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

Zodra u het schemaregister hebt gebruikt om het schema op te slaan, kunt u ernaar verwijzen in de configuratie van de gegevensstroom.

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

Zie Berichtschema's begrijpen voor meer informatie.

Transformatie

Met de transformatiebewerking kunt u de gegevens van de bron transformeren voordat u deze naar de bestemming verzendt. Transformaties zijn optioneel. Als u geen wijzigingen in de gegevens hoeft aan te brengen, neemt u de transformatiebewerking niet op in de configuratie van de gegevensstroom. Meerdere transformaties worden in fasen gekoppeld, ongeacht de volgorde waarin ze zijn opgegeven in de configuratie. De volgorde van de fasen is altijd:

Verrijken: voeg aanvullende gegevens toe aan de brongegevens op basis van een gegevensset en voorwaarde die overeenkomen.
Filter: Filter de gegevens op basis van een voorwaarde.
Een nieuwe eigenschap toewijzen, berekenen, hernoemen of toevoegen: verplaats gegevens van het ene veld naar het andere met een optionele conversie.

Deze sectie is een inleiding tot gegevensstroomtransformaties. Zie Kaartgegevens met behulp van gegevensstromen, Gegevens converteren met behulp van gegevensstroomconversies en Gegevens verrijken met behulp van gegevensstromen voor meer gedetailleerde informatie.

Selecteer in de bewerkingservaring de optie Gegevensstroom>Transformatie toevoegen (optioneel).

Schermopname van bewerkingen om een transformatie toe te voegen aan een gegevensstroom.

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

Verrijken: Referentiegegevens toevoegen

Als u de gegevens wilt verrijken, voegt u eerst een referentiegegevensset toe in het azure IoT Operations-statusarchief. De gegevensset wordt gebruikt om extra gegevens toe te voegen aan de brongegevens op basis van een voorwaarde. De voorwaarde wordt opgegeven als een veld in de brongegevens die overeenkomen met een veld in de gegevensset.

U kunt voorbeeldgegevens laden in het statusarchief met behulp van de CLI van het statusarchief. Sleutelnamen in het statusarchief komen overeen met een gegevensset in de gegevensstroomconfiguratie.

Momenteel wordt de verrijkingsfase niet ondersteund in de bewerkingservaring.

In dit voorbeeld ziet u hoe u het deviceId veld in de brongegevens kunt gebruiken om overeen te komen met het asset veld in de gegevensset:

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

U kunt bijvoorbeeld het deviceId veld in de brongegevens gebruiken om het veld in de gegevensset te vinden asset :

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

Als de gegevensset een record met het asset veld heeft, vergelijkbaar met:

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

De gegevens uit de bron met de deviceId overeenkomende thermostat1 velden bevatten de location velden die manufacturer beschikbaar zijn in filter- en kaartfasen.

Zie Gegevens verrijken met behulp van gegevensstromen en gegevens converteren met behulp van gegevensstromen voor meer informatie over de syntaxis van voorwaarden.

Filter: Gegevens filteren op basis van een voorwaarde

Als u de gegevens op een voorwaarde wilt filteren, kunt u de filter fase gebruiken. De voorwaarde wordt opgegeven als een veld in de brongegevens die overeenkomen met een waarde.

Selecteer Onder Transformeren (optioneel) de optie Filter>toevoegen.
Voer de vereiste instellingen in.

Instelling Beschrijving

Filtervoorwaarde De voorwaarde om de gegevens te filteren op basis van een veld in de brongegevens.

Beschrijving Geef een beschrijving op voor de filtervoorwaarde.

Typ of selecteer ctrl+spatie in @ het filtervoorwaardeveld om gegevenspunten te kiezen in een vervolgkeuzelijst.

U kunt MQTT-metagegevenseigenschappen invoeren met behulp van de indeling @$metadata.user_properties.<property> of @$metadata.topic. U kunt ook $metadata headers invoeren met behulp van de indeling @$metadata.<header>. De $metadata syntaxis is alleen nodig voor MQTT-eigenschappen die deel uitmaken van de berichtkop. Zie veldverwijzingen voor meer informatie.

De voorwaarde kan de velden in de brongegevens gebruiken. U kunt bijvoorbeeld een filtervoorwaarde gebruiken, zoals @temperature > 20 het filteren van gegevens kleiner dan of gelijk aan 20 op basis van het temperatuurveld.
Selecteer Toepassen.

Instelling	Beschrijving
Filtervoorwaarde	De voorwaarde om de gegevens te filteren op basis van een veld in de brongegevens.
Beschrijving	Geef een beschrijving op voor de filtervoorwaarde.

U kunt bijvoorbeeld het temperature veld in de brongegevens gebruiken om de gegevens te filteren:

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

Als het temperature veld groter is dan 20, worden de gegevens doorgegeven aan de volgende fase. Als het temperature veld kleiner is dan of gelijk is aan 20, worden de gegevens gefilterd.

U kunt bijvoorbeeld het temperature veld in de brongegevens gebruiken om de gegevens te filteren:

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

Als het temperature veld groter is dan 20, worden de gegevens doorgegeven aan de volgende fase. Als het temperature veld kleiner is dan of gelijk is aan 20, worden de gegevens gefilterd.

Kaart: Gegevens van het ene veld naar het andere verplaatsen

Als u de gegevens wilt toewijzen aan een ander veld met optionele conversie, kunt u de map bewerking gebruiken. De conversie wordt opgegeven als een formule die gebruikmaakt van de velden in de brongegevens.

In de bewerkingservaring wordt toewijzing momenteel ondersteund met transformaties van compute, naam wijzigen en nieuwe eigenschappen .

Compute

U kunt de compute-transformatie gebruiken om een formule toe te passen op de brongegevens. Deze bewerking wordt gebruikt om een formule toe te passen op de brongegevens en het resultaatveld op te slaan.

Selecteer Onder Transformeren (optioneel) de optie Compute>Add.

Voer de vereiste instellingen in.

Instelling	Beschrijving
Formule selecteren	Kies een bestaande formule in de vervolgkeuzelijst of selecteer Aangepast om handmatig een formule in te voeren.
Uitvoer	Geef de weergavenaam van de uitvoer op voor het resultaat.
Formule	Voer de formule in die moet worden toegepast op de brongegevens.
Beschrijving	Geef een beschrijving op voor de transformatie.
Laatst bekende waarde	Gebruik eventueel de laatst bekende waarde als de huidige waarde niet beschikbaar is.

U kunt een formule invoeren of bewerken in het veld Formule . De formule kan de velden in de brongegevens gebruiken. Typ @ of selecteer Ctrl+ spatiebalk om gegevenspunten te kiezen in een vervolgkeuzelijst.

U kunt MQTT-metagegevenseigenschappen invoeren met behulp van de indeling @$metadata.user_properties.<property> of @$metadata.topic. U kunt ook $metadata headers invoeren met behulp van de indeling @$metadata.<header>. De $metadata syntaxis is alleen nodig voor MQTT-eigenschappen die deel uitmaken van de berichtkop. Zie veldverwijzingen voor meer informatie.

De formule kan de velden in de brongegevens gebruiken. U kunt bijvoorbeeld het temperature veld in de brongegevens gebruiken om de temperatuur naar Celsius te converteren en op te slaan in het temperatureCelsius uitvoerveld.

Selecteer Toepassen.

Naam wijzigen

U kunt de naam van een gegevenspunt wijzigen met behulp van de transformatie Naam wijzigen . Deze bewerking wordt gebruikt om de naam van een gegevenspunt in de brongegevens te wijzigen in een nieuwe naam. De nieuwe naam kan worden gebruikt in de volgende fasen van de gegevensstroom.

Selecteer Onder Transformeren (optioneel) de optie Naam>toevoegen.

Voer de vereiste instellingen in.

Instelling	Beschrijving
Gegevenspunt	Selecteer een gegevenspunt in de vervolgkeuzelijst of voer een $metadata koptekst in.
Nieuwe gegevenspuntnaam	Voer de nieuwe naam in voor het gegevenspunt.
Beschrijving	Geef een beschrijving op voor de transformatie.

Typ @ of selecteer Ctrl+ spatiebalk om gegevenspunten te kiezen in een vervolgkeuzelijst.

Selecteer Toepassen.

Nieuwe eigenschap

U kunt een nieuwe eigenschap toevoegen aan de brongegevens met behulp van de transformatie nieuwe eigenschap . Deze bewerking wordt gebruikt om een nieuwe eigenschap toe te voegen aan de brongegevens. De nieuwe eigenschap kan worden gebruikt in de volgende fasen van de gegevensstroom.

Selecteer onder Transformeren (optioneel) de optie Nieuwe eigenschap>Toevoegen.

Voer de vereiste instellingen in.

Instelling	Beschrijving
Eigenschapssleutel	Voer de sleutel voor de nieuwe eigenschap in.
Eigenschapwaarde	Voer de waarde voor de nieuwe eigenschap in.
Beschrijving	Geef een beschrijving op voor de nieuwe eigenschap.

Selecteer Toepassen.

U hebt toegang tot MQTT-metagegevenseigenschappen met behulp van de indeling $metadata.user_properties.<property> of $metadata.topic. U kunt ook $metadata headers invoeren met behulp van de indeling $metadata.<header>. Zie veldverwijzingen voor meer informatie.

U kunt bijvoorbeeld het temperature veld in de brongegevens gebruiken om de temperatuur naar Celsius te converteren en op te slaan in het temperatureCelsius veld. U kunt de brongegevens ook verrijken met het location veld uit de contextgegevensset:

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

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

Zie Kaartgegevens met behulp van gegevensstromen en Gegevens converteren met behulp van gegevensstromen voor meer informatie.

Gegevens serialiseren volgens een schema

Als u de gegevens wilt serialiseren voordat u deze naar de bestemming verzendt, moet u een schema en serialisatie-indeling opgeven. Anders worden de gegevens geserialiseerd in JSON met de typen die zijn afgeleid. Voor opslageindpunten, zoals Microsoft Fabric of Azure Data Lake, is een schema vereist om gegevensconsistentie te garanderen. Ondersteunde serialisatie-indelingen zijn Parquet en Delta.

Tip

Als u het schema wilt genereren op basis van een voorbeeldgegevensbestand, gebruikt u de Helper voor Schema Gen.

Voor bewerkingen geeft u de schema- en serialisatie-indeling op in de details van het gegevensstroomeindpunt. De eindpunten die serialisatie-indelingen ondersteunen, zijn Microsoft Fabric OneLake, Azure Data Lake Storage Gen 2 en Azure Data Explorer. Als u bijvoorbeeld de gegevens in Delta-indeling wilt serialiseren, moet u een schema uploaden naar het schemaregister en ernaar verwijzen in de eindpuntconfiguratie van de gegevensstroombestemming.

Schermopname van de bewerkingservaring om de serialisatie van het doeleindpunt van de gegevensstroom in te stellen.

Wanneer u een schema uploadt naar het schemaregister, kunt u ernaar verwijzen in de configuratie van de gegevensstroom.

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

Wanneer u een schema uploadt naar het schemaregister, kunt u ernaar verwijzen in de configuratie van de gegevensstroom.

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

Zie Berichtschema's begrijpen voor meer informatie over het schemaregister.

Bestemming

Als u een bestemming voor de gegevensstroom wilt configureren, geeft u de eindpuntverwijzing en het gegevensdoel op. U kunt een lijst met gegevensbestemmingen voor het eindpunt opgeven.

Als u gegevens wilt verzenden naar een andere bestemming dan de lokale MQTT-broker, maakt u een gegevensstroomeindpunt. Zie Eindpunten voor gegevensstromen configureren voor meer informatie. Als de bestemming niet de lokale MQTT-broker is, moet deze worden gebruikt als bron. Zie Gegevensstromen moeten lokaal MQTT-brokereindpunt gebruiken voor meer informatie.

Belangrijk

Voor opslageindpunten is een schema voor serialisatie vereist. Als u een gegevensstroom wilt gebruiken met Microsoft Fabric OneLake, Azure Data Lake Storage, Azure Data Explorer of Lokale opslag, moet u een schemaverwijzing opgeven.

Selecteer het eindpunt van de gegevensstroom dat u als doel wilt gebruiken.

Voor opslageindpunten is een schema voor serialisatie vereist. Als u een Doeleindpunt voor Microsoft Fabric OneLake, Azure Data Lake Storage, Azure Data Explorer of Local Storage kiest, moet u een schemaverwijzing opgeven. Als u bijvoorbeeld de gegevens wilt serialiseren naar een Microsoft Fabric-eindpunt in Delta-indeling, moet u een schema uploaden naar het schemaregister en ernaar verwijzen in de configuratie van het doeleindpunt van de gegevensstroom.
Selecteer Doorgaan om de bestemming te configureren.
Voer de vereiste instellingen voor de bestemming in, inclusief het onderwerp of de tabel waarnaar de gegevens moeten worden verzonden. Zie Gegevensdoel configureren (onderwerp, container of tabel) voor meer informatie.

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

Gegevensbestemming configureren (onderwerp, container of tabel)

Net als bij gegevensbronnen is de gegevensbestemming een concept dat wordt gebruikt om de eindpunten van de gegevensstroom herbruikbaar te houden voor meerdere gegevensstromen. In wezen vertegenwoordigt het de submap in de configuratie van het gegevensstroomeindpunt. Als het eindpunt voor de gegevensstroom bijvoorbeeld een opslageindpunt is, is de gegevensbestemming de tabel in het opslagaccount. Als het eindpunt van de gegevensstroom een Kafka-eindpunt is, is de gegevensbestemming het Kafka-onderwerp.

Eindpunttype	Betekenis van gegevensbestemming	Beschrijving
MQTT (of Event Grid)	Onderwerp	Het MQTT-onderwerp waar de gegevens worden verzonden. Alleen statische onderwerpen worden ondersteund, geen jokertekens.
Kafka (of Event Hubs)	Onderwerp	Het Kafka-onderwerp waar de gegevens worden verzonden. Alleen statische onderwerpen worden ondersteund, geen jokertekens. Als het eindpunt een Event Hubs-naamruimte is, is de gegevensbestemming de afzonderlijke Event Hub binnen de naamruimte.
Azure Data Lake Storage	Container	De container in het opslagaccount. Niet de tabel.
Microsoft Fabric OneLake	Tabel of map	Komt overeen met het geconfigureerde padtype voor het eindpunt.
Azure Data Explorer	Tabel	De tabel in de Azure Data Explorer-database.
Lokale opslag	Map	De map- of mapnaam in de permanente volumekoppeling van de lokale opslag. Wanneer u Azure Container Storage gebruikt dat is ingeschakeld door Azure Arc Cloud Ingest Edge Volumes, moet dit overeenkomen met de `spec.path` parameter voor het subvolume dat u hebt gemaakt.

De gegevensbestemming configureren:

Wanneer u de bewerkingservaring gebruikt, wordt het gegevensdoelveld automatisch geïnterpreteerd op basis van het eindpunttype. Als het eindpunt voor de gegevensstroom bijvoorbeeld een opslageindpunt is, wordt u op de pagina met doelgegevens gevraagd om de containernaam in te voeren. Als het eindpunt van de gegevensstroom een MQTT-eindpunt is, wordt u op de pagina met doelgegevens gevraagd het onderwerp in te voeren, enzovoort.

Schermopname van de bewerkingservaring waarin de gebruiker wordt gevraagd een MQTT-onderwerp in te voeren op basis van het eindpunttype.

De syntaxis is hetzelfde voor alle gegevensstroomeindpunten:

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

Als u bijvoorbeeld gegevens wilt terugsturen naar de lokale MQTT-broker, gebruikt u de volgende configuratie:

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

Als u een aangepast Event Hub-eindpunt hebt, ziet de configuratie er dan als volgt uit:

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

Een ander voorbeeld van het gebruik van een opslageindpunt als de bestemming:

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

De syntaxis is hetzelfde voor alle gegevensstroomeindpunten:

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

Als u bijvoorbeeld gegevens wilt terugsturen naar de lokale MQTT-broker, gebruikt u de volgende configuratie:

destinationSettings:
  endpointRef: default
  dataDestination: example-topic

Als u een aangepast Event Hub-eindpunt hebt, ziet de configuratie er dan als volgt uit:

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

Een ander voorbeeld van het gebruik van een opslageindpunt als de bestemming:

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

Opmerking

Het volgende voorbeeld is een gegevensstroomconfiguratie die gebruikmaakt van het MQTT-eindpunt voor de bron en het doel. De bron filtert de gegevens uit het MQTT-onderwerp azure-iot-operations/data/thermostat. De transformatie converteert de temperatuur naar Fahrenheit en filtert de gegevens waarbij de temperatuur vermenigvuldigd met de vochtigheid kleiner is dan 100000. De bestemming verzendt de gegevens naar het MQTT-onderwerp factory.

Zie bicep- of Kubernetes-tabbladen voor het configuratievoorbeeld.

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 dataflow endpoint
resource defaultDataflowEndpoint 'Microsoft.IoTOperations/instances/dataflowEndpoints@2024-11-01' existing = {
  parent: aioInstance
  name: 'default'
}

// Pointer to the default dataflow 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 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'
        }
      }
    ]
  }
}

apiVersion: connectivity.iotoperations.azure.com/v1
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

Zie Azure REST API - Dataflow en de quickstart Bicep voor meer voorbeelden van gegevensstroomconfiguraties.

Controleren of een gegevensstroom werkt

Volg de zelfstudie: Bidirectionele MQTT-brug naar Azure Event Grid om te controleren of de gegevensstroom werkt.

Configuratie van gegevensstroom exporteren

Als u de configuratie van de gegevensstroom wilt exporteren, kunt u de bewerkingservaring gebruiken of door de aangepaste gegevensstroomresource te exporteren.

Selecteer de gegevensstroom die u wilt exporteren en selecteer Exporteren op de werkbalk.

Schermopname van het gebruik van bewerkingen om een gegevensstroom te exporteren.

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

Juiste gegevensstroomconfiguratie

Controleer het volgende om ervoor te zorgen dat de gegevensstroom werkt zoals verwacht:

Het standaard MQTT-gegevensstroomeindpunt moet worden gebruikt als de bron of het doel.
Het gegevensstroomprofiel bestaat en wordt verwezen in de configuratie van de gegevensstroom.
De bron is een MQTT-eindpunt, Kafka-eindpunt of een asset. Eindpunten van het opslagtype kunnen niet worden gebruikt als bron.
Wanneer u Event Grid als bron gebruikt, wordt het aantal exemplaren van het gegevensstroomprofiel ingesteld op 1 omdat Event Grid MQTT Broker geen ondersteuning biedt voor gedeelde abonnementen.
Wanneer u Event Hubs als bron gebruikt, is elke Event Hub in de naamruimte een afzonderlijk Kafka-onderwerp en moet deze worden opgegeven als de gegevensbron.
Transformatie, indien gebruikt, wordt geconfigureerd met de juiste syntaxis, inclusief het juiste ontsnappen van speciale tekens.
Wanneer u eindpunten van het opslagtype als bestemming gebruikt, wordt een schema opgegeven.

Delen via

Gegevensstromen configureren in Azure IoT-bewerkingen

Vereisten

Gegevensstroomprofiel

Gegevensstroomeindpunten

Aan de slag

Bron

Optie 1: Standaard MQTT-eindpunt gebruiken als bron

Optie 2: Asset als bron gebruiken

Optie 3: Aangepast MQTT- of Kafka-gegevensstroomeindpunt als bron gebruiken

Gegevensbronnen (MQTT- of Kafka-onderwerpen) configureren

MQTT-onderwerpen

Gedeelde abonnementen

Kafka-onderwerpen

Bronschema opgeven

Transformatie

Verrijken: Referentiegegevens toevoegen

Filter: Gegevens filteren op basis van een voorwaarde

Kaart: Gegevens van het ene veld naar het andere verplaatsen

Compute

Naam wijzigen

Nieuwe eigenschap

Gegevens serialiseren volgens een schema

Bestemming

Gegevensbestemming configureren (onderwerp, container of tabel)

Opmerking

Controleren of een gegevensstroom werkt

Configuratie van gegevensstroom exporteren

Juiste gegevensstroomconfiguratie

Volgende stappen

Feedback

Aanvullende resources