Konfigurace toků dat v operacích Azure IoT

Článek
11/02/2024

Důležité

Azure IoT Operations Preview – Služba Azure Arc je aktuálně ve verzi Preview. Tento software ve verzi Preview byste neměli používat v produkčních prostředích.

Až bude dostupná obecně dostupná verze, budete muset nasadit novou instalaci operací Azure IoT. Nebudete moct upgradovat instalaci verze Preview.

Právní podmínky, které se vztahují na funkce Azure, které jsou ve verzi beta, ve verzi Preview nebo které ještě nejsou vydány v obecné dostupnosti, najdete v dodatečných podmínkách použití pro verze Microsoft Azure Preview.

Tok dat je cesta, kterou data přebíná ze zdroje do cíle s volitelnými transformacemi. Tok dat můžete nakonfigurovat vytvořením vlastního prostředku toku dat nebo pomocí portálu Azure IoT Operations Studio. Tok dat se skládá ze tří částí: zdroje, transformace a cíle.

Diagram toku dat znázorňující tok ze zdroje k transformaci a následnému cíli

Pokud chcete definovat zdroj a cíl, musíte nakonfigurovat koncové body toku dat. Transformace je volitelná a může zahrnovat operace, jako je rozšiřování dat, filtrování dat a mapování dat na jiné pole.

Důležité

Každý tok dat musí mít jako zdroj nebo cíl výchozí koncový bod zprostředkovatele MQTT operations Azure IoT Operations.

K vytvoření toku dat můžete použít provozní prostředí v operacích Azure IoT. Provozní prostředí poskytuje vizuální rozhraní pro konfiguraci toku dat. Pomocí Bicep můžete také vytvořit tok dat pomocí souboru šablony Bicep nebo pomocí Kubernetes vytvořit tok dat pomocí souboru YAML.

Pokračujte ve čtení a zjistěte, jak nakonfigurovat zdroj, transformaci a cíl.

Požadavky

Toky dat můžete nasadit hned, jak máte instanci Azure IoT Operations Preview pomocí výchozího profilu a koncového bodu toku dat. Můžete ale chtít nakonfigurovat profily toku dat a koncové body pro přizpůsobení toku dat.

Profil toku dat

Profil toku dat určuje počet instancí pro toky dat, které se mají použít. Pokud nepotřebujete více skupin toků dat s různými nastaveními škálování, můžete použít výchozí profil toku dat. Informace o konfiguraci profilu toku dat najdete v tématu Konfigurace profilů toku dat.

Koncové body toku dat

Koncové body toku dat jsou potřeba ke konfiguraci zdroje a cíle toku dat. Abyste mohli rychle začít, můžete použít výchozí koncový bod toku dat pro místního zprostředkovatele MQTT. Můžete také vytvořit další typy koncových bodů toku dat, jako jsou Kafka, Event Hubs nebo Azure Data Lake Storage. Informace o konfiguraci jednotlivých typů koncového bodu toku dat najdete v tématu Konfigurace koncových bodů toku dat.

Začínáme

Jakmile budete mít požadavky, můžete začít vytvářet tok dat.

Pokud chcete vytvořit tok dat v provozním prostředí, vyberte Vytvořit tok> dat. Pak uvidíte stránku, kde můžete nakonfigurovat zdroj, transformaci a cíl toku dat.

Snímek obrazovky s provozním prostředím k vytvoření toku dat

Vytvořte soubor Bicep .bicep , který začne vytvářet tok dat. Tento příklad ukazuje strukturu toku dat obsahujícího konfiguraci zdroje, transformace a cíle.

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

Vytvořte soubor manifestu .yaml Kubernetes, abyste mohli začít vytvářet tok dat. Tento příklad ukazuje strukturu toku dat obsahujícího konfiguraci zdroje, transformace a cíle.

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

V následujících částech se dozvíte, jak nakonfigurovat typy operací toku dat.

Zdroj

Pokud chcete nakonfigurovat zdroj pro tok dat, zadejte odkaz na koncový bod a seznam zdrojů dat pro koncový bod.

Použití prostředku jako zdroje

Prostředek můžete použít jako zdroj pro tok dat. Použití prostředku jako zdroje je k dispozici pouze v provozním prostředí.

V části Podrobnosti o zdroji vyberte Asset.
Vyberte prostředek, který chcete použít jako zdrojový koncový bod.
Vyberte Pokračovat.

Zobrazí se seznam datových bodů pro vybraný prostředek.
Vyberte Použít , pokud chcete prostředek použít jako zdrojový koncový bod.

Při použití prostředku jako zdroje se definice prostředku použije k odvození schématu pro tok dat. Definice prostředku obsahuje schéma datových bodů prostředku. Další informace najdete v tématu Vzdálená správa konfigurací prostředků.

Po nakonfigurování data z prostředku dosáhla toku dat přes místního zprostředkovatele MQTT. Proto při použití prostředku jako zdroje používá tok dat jako skutečný zdroj výchozí koncový bod zprostředkovatele MQTT.

Použití výchozího koncového bodu MQTT jako zdroje

V části Podrobnosti o zdroji vyberte MQTT.

Zadejte následující nastavení pro zdroj MQTT:

Nastavení	Popis
Téma MQTT	Filtr tématu MQTT pro přihlášení k odběru příchozích zpráv. Viz Témata týkající se konfigurace MQTT nebo Kafka.
Schéma zpráv	Schéma, které se má použít k deserializaci příchozích zpráv. Viz Určení schématu pro deserializaci dat.

Vyberte Použít.

Koncový bod MQTT je nakonfigurovaný v souboru šablony Bicep. Například následující koncový bod je zdrojem toku dat.

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

dataSources Tady můžete zadat několik témat MQTT nebo Kafka, aniž byste museli upravovat konfiguraci koncového bodu. Tato flexibilita znamená, že stejný koncový bod je možné znovu použít napříč několika toky dat, i když se témata liší. Další informace najdete v tématu Konfigurace zdrojů dat.

Pokud chcete například nakonfigurovat zdroj pomocí koncového bodu MQTT a dvou filtrů témat MQTT, použijte následující konfiguraci:

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

Protože dataSources umožňuje zadat témata MQTT nebo Kafka beze změny konfigurace koncového bodu, můžete znovu použít koncový bod pro více toků dat, i když se témata liší. Další informace najdete v tématu Konfigurace zdrojů dat.

Pokud se jako zdroj nepoužívá výchozí koncový bod, musí se použít jako cíl. Další informace najdete v tématu Toky dat musí používat místní koncový bod zprostředkovatele MQTT.

Použití vlastního koncového bodu toku dat Kafka nebo MQTT jako zdroje

Pokud jste vytvořili vlastní koncový bod MQTT nebo toku dat Kafka (například pro použití se službou Event Grid nebo Event Hubs), můžete ho použít jako zdroj pro tok dat. Mějte na paměti, že koncové body typu úložiště, jako jsou Data Lake nebo Fabric OneLake, se nedají použít jako zdroj.

Ke konfiguraci použijte Kubernetes YAML nebo Bicep. Nahraďte zástupné hodnoty vlastním názvem a tématy koncového bodu.

Použití vlastního koncového bodu MQTT nebo Kafka jako zdroje se v současné době v provozním prostředí nepodporuje.

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

Konfigurace zdrojů dat (témata MQTT nebo Kafka)

Ve zdroji můžete zadat několik témat MQTT nebo Kafka, aniž byste museli upravovat konfiguraci koncového bodu toku dat. Tato flexibilita znamená, že stejný koncový bod je možné znovu použít napříč několika toky dat, i když se témata liší. Další informace najdete v tématu Opětovné použití koncových bodů toku dat.

Témata MQTT

Pokud je zdrojem koncový bod MQTT (zahrnutý event Grid), můžete k odběru příchozích zpráv použít filtr tématu MQTT. Filtr témat může obsahovat zástupné cardy pro přihlášení k odběru více témat. Například thermostats/+/telemetry/temperature/# se přihlásí k odběru všech zpráv telemetrie teploty z termostatů. Konfigurace filtrů témat MQTT:

V podrobnostech o zdroji toku dat v provozním prostředí vyberte MQTT a pak pomocí pole tématu MQTT určete filtr tématu MQTT pro přihlášení k odběru příchozích zpráv.

Poznámka:

V provozním prostředí je možné zadat pouze jeden filtr témat MQTT. Pokud chcete použít více filtrů témat MQTT, použijte Bicep nebo Kubernetes.

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

Příklad s několika filtry témat MQTT se zástupnými cardy:

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

Zástupný znak + se zde používá k výběru všech zařízení v rámci témat thermostats a humidifiers témat. Zástupný # znak slouží k výběru všech zpráv telemetrie pod všemi dílčími tématy a humidity tématytemperature.

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

Příklad s několika filtry témat MQTT se zástupnými cardy:

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

Sdílená předplatná

Pokud chcete používat sdílená předplatná se zdroji MQTT, můžete zadat téma sdíleného předplatného $shared/<GROUP_NAME>/<TOPIC_FILTER>ve formě .

V provozním prostředí se zobrazí podrobnosti o zdroji toku dat, vyberte MQTT a pomocí pole téma MQTT určete skupinu a téma sdíleného předplatného.

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

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

Pokud je počet instancí v profilu toku dat větší než 1, je sdílené předplatné automaticky povolené pro všechny toky dat, které používají zdroj MQTT. V tomto případě $shared se přidá předpona a automaticky se vygeneruje název skupiny sdílených předplatných. Pokud máte například profil toku dat s počtem instancí 3 a váš tok dat jako zdroj nakonfigurovaný s tématy topic1 používá koncový bod MQTT a topic2automaticky se převedou na sdílená předplatná jako $shared/<GENERATED_GROUP_NAME>/topic1 a $shared/<GENERATED_GROUP_NAME>/topic2. Pokud chcete použít jiné ID skupiny sdílených předplatných, můžete ho přepsat v tématu, například $shared/mygroup/topic1.

Důležité

Toky dat vyžadují sdílené předplatné, pokud je počet instancí větší než 1, je důležité při použití zprostředkovatele Event Grid MQTT jako zdroje, protože nepodporuje sdílená předplatná. Pokud se chcete vyhnout chybějícím zprávám, nastavte počet instancí profilu toku dat na hodnotu 1 při použití zprostředkovatele Event Grid MQTT jako zdroje. To znamená, že tok dat je odběratel a přijímá zprávy z cloudu.

Témata Kafka

Pokud je zdrojem koncový bod Kafka (zahrnutá služba Event Hubs), zadejte jednotlivá témata kafka, která se mají přihlásit k odběru příchozích zpráv. Zástupné cardy nejsou podporovány, takže je nutné zadat každé téma staticky.

Poznámka:

Při použití služby Event Hubs prostřednictvím koncového bodu Kafka je každé jednotlivé centrum událostí v rámci oboru názvů téma Kafka. Pokud máte například obor názvů služby Event Hubs se dvěma centry událostí a thermostats humidifiersmůžete každé centrum událostí zadat jako téma Kafka.

Konfigurace témat Kafka:

Použití koncového bodu Kafka jako zdroje se v současné době nepodporuje v provozním prostředí.

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

Určení schématu pro deserializaci dat

Pokud zdrojová data obsahují volitelná pole nebo pole s různými typy, zadejte schéma deserializace, které zajistí konzistenci. Data můžou mít například pole, která nejsou ve všech zprávách. Bez schématu transformace nemůže zpracovat tato pole, protože by měly prázdné hodnoty. Pomocí schématu můžete zadat výchozí hodnoty nebo pole ignorovat.

Určení schématu je relevantní pouze při použití zdroje MQTT nebo Kafka. Pokud je zdrojem prostředek, schéma se automaticky odvodí z definice prostředku.

Konfigurace schématu použitého k deserializaci příchozích zpráv ze zdroje:

V provozním prostředí se zobrazí podrobnosti o zdroji toku dat, vyberte MQTT a pomocí pole Schéma zprávy určete schéma. Soubor schématu můžete nahrát napřed pomocí tlačítka Nahrát . Další informace najdete v tématu Vysvětlení schémat zpráv.

Jakmile použijete registr schématu k uložení schématu, můžete na něj odkazovat v konfiguraci toku dat.

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

Jakmile použijete registr schématu k uložení schématu, můžete na něj odkazovat v konfiguraci toku dat.

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

Transformace

Transformační operace je místo, kde můžete transformovat data ze zdroje předtím, než je odešlete do cíle. Transformace jsou volitelné. Pokud nepotřebujete provádět změny dat, nezahrnujte do konfigurace toku dat operaci transformace. Více transformací je zřetězených ve fázích bez ohledu na pořadí, ve kterém jsou zadané v konfiguraci. Pořadí fází je vždy:

Obohacení, přejmenování nebo přidání nové vlastnosti: Přidání dalších dat do zdrojových dat zadaných datovou sadou a podmínkou, které se mají shodovat.
Filtr: Filtruje data na základě podmínky.
Mapování nebo výpočty: Přesun dat z jednoho pole do druhého pomocí volitelného převodu

V provozním prostředí vyberte Přidat transformaci toku>dat (volitelné).

Snímek obrazovky s provozním prostředím pro přidání transformace do toku dat

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

Obohacení: Přidání referenčních dat

K obohacení dat můžete použít referenční datovou sadu v úložišti distribuovaných stavů Azure IoT Operations (DSS). Datová sada se používá k přidání dalších dat do zdrojových dat na základě podmínky. Podmínka se zadává jako pole ve zdrojových datech, které odpovídá poli v datové sadě.

Ukázková data můžete načíst do DSS pomocí ukázky nástroje sady DSS. Názvy klíčů v úložišti distribuovaných stavů odpovídají datové sadě v konfiguraci toku dat.

V provozním prostředí se fáze rozšiřování v současné době podporuje pomocí transformací vlastnosti Rename a New .

V provozním prostředí vyberte tok dat a pak přidejte transformaci (volitelné).
Zvolte Přejmenovat nebo Nová transformace vlastností a pak vyberte Přidat.

Tento příklad ukazuje, jak můžete pole ve zdrojových datech použít deviceId ke spárování asset pole v datové sadě:

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

Můžete například použít deviceId pole ve zdrojových datech tak, aby odpovídalo asset poli v datové sadě:

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

Pokud datová sada obsahuje záznam s polem asset , podobá se:

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

Data ze zdroje s odpovídajícími deviceId thermostat1 poli mají location pole a manufacturer pole dostupná ve fázích filtrování a mapování.

Další informace o syntaxi podmínky naleznete v tématu Obohacení dat pomocí toků dat a převod dat pomocí toků dat.

Filtr: Filtrování dat na základě podmínky

K filtrování dat v podmínce můžete použít filter fázi. Podmínka je určena jako pole ve zdrojových datech, které odpovídá hodnotě.

V části Transformace (volitelné) vyberte Přidat filtr>.
Zvolte datové body, které chcete zahrnout do datové sady.
Přidejte podmínku filtru a popis.
Vyberte Použít.

Můžete například použít podmínku filtru, jako temperature > 20 je filtrování dat menší než nebo rovno 20 na základě pole teploty.

Pole ve zdrojových datech můžete například použít temperature k filtrování dat:

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

temperature Pokud je pole větší než 20, předají se data do další fáze. temperature Pokud je pole menší nebo rovno 20, data se filtrují.

Pole ve zdrojových datech můžete například použít temperature k filtrování dat:

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

temperature Pokud je pole větší než 20, předají se data do další fáze. temperature Pokud je pole menší nebo rovno 20, data se filtrují.

Mapa: Přesun dat z jednoho pole do druhého

Pokud chcete data namapovat na jiné pole s volitelným převodem, můžete tuto operaci použít map . Převod je určen jako vzorec, který používá pole ve zdrojových datech.

V provozním prostředí se v současné době podporuje mapování pomocí transformací Compute .

V části Transformace (volitelné) vyberte Přidat výpočetní prostředky>.
Zadejte požadovaná pole a výrazy.
Vyberte Použít.

Pole ve zdrojových datech můžete například použít temperature k převodu teploty na stupně Celsia a jeho uložení do temperatureCelsius pole. Zdrojová data můžete také rozšířit o location pole z kontextové datové sady:

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

Další informace najdete v tématu Mapování dat pomocí toků dat a převodu dat pomocí toků dat.

Serializace dat podle schématu

Pokud chcete data před odesláním do cíle serializovat, musíte zadat schéma a formát serializace. V opačném případě se data serializují ve formátu JSON s odvozenými typy. Koncové body úložiště, jako jsou Microsoft Fabric nebo Azure Data Lake, vyžadují schéma pro zajištění konzistence dat. Podporované formáty serializace jsou Parquet a Delta.

V současné době není v provozním prostředí podporováno určení výstupního schématu a serializace.

Po nahrání schématu do registru schématu na něj můžete odkazovat v konfiguraci toku dat.

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

Po nahrání schématu do registru schématu na něj můžete odkazovat v konfiguraci toku dat.

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

Další informace o registru schématu naleznete v tématu Vysvětlení schémat zpráv.

Cíl

Pokud chcete nakonfigurovat cíl pro tok dat, zadejte odkaz na koncový bod a cíl dat. Můžete zadat seznam cílů dat pro koncový bod.

Pokud chcete odesílat data do jiného cíle, než je místní zprostředkovatel MQTT, vytvořte koncový bod toku dat. Postup najdete v tématu Konfigurace koncových bodů toku dat. Pokud cíl není místním zprostředkovatelem MQTT, musí se použít jako zdroj. Další informace najdete v tématu Toky dat musí používat místní koncový bod zprostředkovatele MQTT.

Důležité

Koncové body úložiště vyžadují odkaz na schéma. Pokud jste vytvořili cílové koncové body úložiště pro Microsoft Fabric OneLake, ADLS Gen2, Azure Data Explorer a Local Storage, musíte zadat odkaz na schéma.

Vyberte koncový bod toku dat, který chcete použít jako cíl.
Vyberte Pokračovat a nakonfigurujte cíl.
Zadejte požadovaná nastavení cíle, včetně tématu nebo tabulky pro odeslání dat. Další informace najdete v tématu Konfigurace cíle dat (téma, kontejner nebo tabulka).

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

Konfigurace cíle dat (téma, kontejner nebo tabulka)

Podobně jako u zdrojů dat je cílem dat koncept, který slouží k opakovanému použití koncových bodů toku dat napříč několika toky dat. V podstatě představuje podadresář v konfiguraci koncového bodu toku dat. Pokud je například koncovým bodem toku dat koncový bod úložiště, cílem dat je tabulka v účtu úložiště. Pokud je koncový bod toku dat koncovým bodem Kafka, je cílem dat téma Kafka.

Typ koncového bodu	Význam cíle dat	Popis
MQTT (nebo Event Grid)	Téma	Téma MQTT, ve kterém se data odesílají. Podporují se pouze statická témata, žádná zástupná dokumentace.
Kafka (nebo Event Hubs)	Téma	Téma Kafka, ve kterém se data odesílají. Podporují se pouze statická témata, žádná zástupná dokumentace. Pokud je koncový bod oborem názvů služby Event Hubs, je cílem dat jednotlivé centrum událostí v rámci oboru názvů.
Azure Data Lake Storage	Kontejner	Kontejner v účtu úložiště. Ne tabulka.
Microsoft Fabric OneLake	Tabulka nebo složka	Odpovídá nakonfigurovaného typu cesty pro koncový bod.
Průzkumník dat Azure	Table	Tabulka v databázi Azure Data Exploreru.
Lokální úložiště	Složka	Název složky nebo adresáře v připojení trvalého svazku místního úložiště. Pokud používáte Službu Azure Container Storage povolenou svazky Azure Arc Cloud Ingest Edge, musí se shodovat s `spec.path` parametrem pro podvolum, který jste vytvořili.

Konfigurace cíle dat:

Při použití provozního prostředí se cílové pole dat automaticky interpretuje na základě typu koncového bodu. Pokud je například koncový bod toku dat koncovým bodem úložiště, zobrazí se na stránce podrobností cíle výzva k zadání názvu kontejneru. Pokud je koncový bod toku dat koncovým bodem MQTT, zobrazí se na stránce podrobností cíle výzva k zadání tématu atd.

Snímek obrazovky znázorňující provozní prostředí s výzvou, aby uživatel zadal téma MQTT s daným typem koncového bodu

Syntaxe je stejná pro všechny koncové body toku dat:

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

Pokud například chcete odeslat data zpět do místního zprostředkovatele MQTT, použijte statické téma MQTT, použijte následující konfiguraci:

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

Pokud máte vlastní koncový bod centra událostí, konfigurace by vypadala takto:

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

Další příklad použití koncového bodu úložiště jako cíle:

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

Syntaxe je stejná pro všechny koncové body toku dat:

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

Pokud například chcete odeslat data zpět do místního zprostředkovatele MQTT, použijte statické téma MQTT, použijte následující konfiguraci:

destinationSettings:
  endpointRef: default
  dataDestination: example-topic

Pokud máte vlastní koncový bod centra událostí, konfigurace by vypadala takto:

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

Další příklad použití koncového bodu úložiště jako cíle:

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

Příklad

Následující příklad je konfigurace toku dat, která používá koncový bod MQTT pro zdroj a cíl. Zdroj filtruje data z tématu azure-iot-operations/data/thermostatMQTT . Transformace převede teplotu na Fahrenheita a vyfiltruje data, ve kterých je teplota vynásobená vlhkostí menší než 1 00000. Cíl odešle data do tématu factoryMQTT .

Příklad konfigurace najdete na kartách Bicep nebo Kubernetes.

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

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

Další příklady konfigurací toku dat najdete v tématu Azure REST API – Tok dat a rychlý start Bicep.

Ověření, že tok dat funguje

Postupujte podle kurzu: Obousměrný most MQTT do Azure Event Gridu a ověřte, že tok dat funguje.

Export konfigurace toku dat

Pokud chcete exportovat konfiguraci toku dat, můžete použít provozní prostředí nebo exportovat vlastní prostředek toku dat.

Vyberte tok dat, který chcete exportovat, a na panelu nástrojů vyberte Exportovat .

Snímek obrazovky s provozním prostředím k exportu toku dat

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

Správná konfigurace toku dat

Pokud chcete zajistit, aby tok dat fungoval podle očekávání, ověřte následující:

Výchozí koncový bod toku dat MQTT musí být použit jako zdroj nebo cíl.
Profil toku dat existuje a odkazuje se na ho v konfiguraci toku dat.
Zdrojem je koncový bod MQTT, koncový bod Kafka nebo prostředek. Koncové body typu úložiště nejde použít jako zdroj.
Pokud jako zdroj používáte Event Grid, nastaví se počet instancí profilu toku dat na 1, protože zprostředkovatel Event Grid MQTT nepodporuje sdílená předplatná.
Při použití služby Event Hubs jako zdroje je každé centrum událostí v oboru názvů samostatné téma Kafka a musí být zadáno jako zdroj dat.
Transformace, pokud se používá, je nakonfigurována se správnou syntaxí, včetně správného uvozování speciálních znaků.
Při použití koncových bodů typu úložiště jako cíle se zadává schéma.

Sdílet prostřednictvím

Konfigurace toků dat v operacích Azure IoT

Požadavky

Profil toku dat

Koncové body toku dat

Začínáme

Zdroj

Použití prostředku jako zdroje

Použití výchozího koncového bodu MQTT jako zdroje

Použití vlastního koncového bodu toku dat Kafka nebo MQTT jako zdroje

Konfigurace zdrojů dat (témata MQTT nebo Kafka)

Témata MQTT

Sdílená předplatná

Témata Kafka

Určení schématu pro deserializaci dat

Transformace

Obohacení: Přidání referenčních dat

Filtr: Filtrování dat na základě podmínky

Mapa: Přesun dat z jednoho pole do druhého

Serializace dat podle schématu

Cíl

Konfigurace cíle dat (téma, kontejner nebo tabulka)

Příklad

Ověření, že tok dat funguje

Export konfigurace toku dat

Správná konfigurace toku dat

Další kroky

Váš názor

Další materiály