Vytváření toků dat v operacích Azure IoT

Důležité

Tato stránka obsahuje pokyny ke správě komponent operací Azure IoT pomocí manifestů nasazení Kubernetes, které jsou ve verzi Preview. Tato funkce je poskytována s několika omezeními a neměla by se používat pro produkční úlohy.

Právní podmínky, které platí pro funkce Azure, které jsou ve verzi beta, verzi Preview nebo které zatím nejsou veřejně dostupné, najdete v Dodatečných podmínkách použití pro Microsoft Azure verze Preview.

Tok dat je trasa, kterou data přebírá ze zdroje do cíle s volitelnými transformacemi. Tok dat můžete nakonfigurovat vytvořením vlastní prostředku Data flow nebo pomocí webového uživatelského rozhraní operačního prostředí. 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 výchozí koncový bod buď místního MQTT zprostředkovatele služby Azure IoT Operations jako zdroj nebo cíl.

Provozní prostředí v operacích Azure IoT můžete použít k vytvoření toku dat. Provozní prostředí poskytuje vizuální rozhraní pro konfiguraci toku dat. Pomocí Bicep můžete také vytvořit tok dat pomocí souboru 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, jakmile budete mít instanci Azure IoT Operations, a to pomocí výchozího profilu toku dat a koncového bodu. Můžete ale chtít nakonfigurovat profily toku dat a koncové body pro přizpůsobení toku dat.

Profil toku dat

Pokud pro toky dat nepotřebujete různá nastavení škálování, použijte výchozí profil toku dat poskytovaný operacemi Azure IoT. Neměli byste přidružovat příliš mnoho toků dat k jednomu profilu toku dat. Pokud máte velký počet toků dat, rozdělte je do několika profilů toků dat, abyste snížili riziko překročení limitu velikosti konfigurace profilu toku dat 70.

Informace o konfiguraci nového 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, OpenTelemetry nebo Azure Data Lake Storage. Informace o konfiguraci jednotlivých typů koncových bodů toku dat najdete v tématu Konfigurace koncových bodů toku dat.

Začínáme

Jakmile budete mít potřebné předpoklady, můžete začít vytvářet tok dat.

Pokud chcete vytvořit tok dat v operations experience, vyberte Tok dat>Vytvořit tok dat.
Vyberte zástupný název new-data-flow a nastavte vlastnosti toku dat. Zadejte název toku dat a zvolte profil toku dat, který chcete použít. Výchozí profil toku dat je ve výchozím nastavení vybraný. Další informace o profilech toku dat najdete v tématu Konfigurace profilu toku dat.

Důležité

Profil toku dat můžete zvolit jenom při vytváření toku dat. Po vytvoření toku dat nemůžete změnit profil toku dat. Pokud chcete změnit profil toku dat existujícího toku dat, odstraňte původní tok dat a vytvořte nový s novým profilem toku dat.
Nakonfigurujte zdrojový, transformační a cílový koncový bod toku dat tak, že vyberete položky v diagramu toku dat.

Příkaz az iot ops dataflow apply použijte k vytvoření nebo změně toku dat.

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

Parametr --config-file je cesta a název souboru JSON, který obsahuje vlastnosti prostředku.

V tomto příkladu předpokládejme konfigurační soubor s následujícím data-flow.json obsahem uloženým v domovském adresáři uživatele:

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

Tady je ukázkový příkaz pro vytvoření nebo aktualizaci toku dat pomocí výchozího profilu toku dat:

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

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-11-01' existing = {
  name: aioInstanceName
}

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

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

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

resource dataflow 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflows@2024-11-01' = {
  // Reference to the parent data flow profile, the default profile in this case
  // Same usage as profileRef in Kubernetes YAML
  parent: defaultDataflowProfile
  name: dataflowName
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    mode: 'Enabled'
    operations: [
      {
        operationType: 'Source'
        sourceSettings: {
          // See source configuration section
        }
      }
      // Transformation optional
      {
        operationType: 'BuiltInTransformation'
        builtInTransformationSettings: {
          // See transformation configuration section
        }
      }
      {
        operationType: 'Destination'
        destinationSettings: {
          // See destination configuration section
        }
      }
    ]
  }
}

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/v1
kind: Dataflow
metadata:
  name: <DATAFLOW_NAME>
  namespace: azure-iot-operations
spec:
  # Reference to the default data flow profile
  # This field is required when configuring via Kubernetes YAML
  # The syntax is different when using Bicep
  profileRef: default 
  mode: Enabled
  operations:
    - operationType: Source
      sourceSettings:
        # See source configuration section
      # Transformation optional
    - operationType: BuiltInTransformation
      builtInTransformationSettings:
        # See transformation configuration section
    - operationType: Destination
      destinationSettings:
        # See destination configuration section

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. Jako zdroj toku dat vyberte jednu z následujících možností.

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

Možnost 1: Použití výchozího koncového bodu zprostředkovatele zpráv jako zdroje

V části Podrobnosti o zdroji vyberte zprostředkovatele zpráv.

Zadejte následující nastavení pro zdroj zprostředkovatele zpráv:

Nastavení	Popis
Koncový bod toku dat	Vyberte výchozí nastavení pro použití výchozího koncového bodu zprostředkovatele zpráv MQTT.
Téma	Filtr tématu pro přihlášení k odběru příchozích zpráv. Použijte Téma (Témata)>Přidat řádek pro přidání více témat. Další informace o tématech najdete v tématu Konfigurace témat 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.

Tady je příklad konfigurace zdrojového koncového bodu pro výchozí koncový bod zprostředkovatele MQTT:

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

Koncový bod zprostředkovatele zpráv je nakonfigurován v souboru Bicep. Například následující koncový bod je zdrojem toku dat.

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

Pokud chcete například nakonfigurovat zdroj pomocí koncového bodu zprostředkovatele zpráv a dva filtry témat, použijte následující konfiguraci:

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

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

Možnost 2: Použití materiálu jako zdroje

Jako zdroj pro tok dat můžete použít prostředek. Použití assetu jako zdroje je k dispozici pouze v operační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 toku 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í se data ze zařízení dostanou do toku dat prostřednictvím místního zprostředkovatele MQTT. Při použití aktiva jako zdroje ve skutečnosti tok dat používá výchozí koncový bod místního zprostředkovatele MQTT jako zdroj.

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

Pokud jste vytvořili vlastní koncový bod toku dat MQTT nebo Kafka (například pro použití se službou Event Grid nebo Event Hubs), můžete ho použít jako zdroj toku 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.

V části Podrobnosti o zdroji vyberte zprostředkovatele zpráv.

Zadejte následující nastavení pro zdroj zprostředkovatele zpráv:

Nastavení	Popis
Koncový bod toku dat	Pomocí tlačítka Pro výběr vyberte vlastní koncový bod toku dat MQTT nebo Kafka. Další informace najdete v tématu Konfigurace koncových bodů toku dat MQTT nebo konfigurace koncových bodů toku dat Azure Event Hubs a Kafka.
Téma	Filtr tématu pro přihlášení k odběru příchozích zpráv. Použijte Téma (Témata)>Přidat řádek pro přidání více témat. Další informace o tématech najdete v tématu Konfigurace témat 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.

Nahraďte zástupné hodnoty vlastním názvem koncového bodu a svými tématy.

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

Nahraďte zástupné hodnoty vlastním názvem koncového bodu a svými tématy.

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

Nahraďte zástupné hodnoty vlastním názvem koncového bodu a svými tématy.

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 více 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é opakovaně používat napříč více 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é znaky pro přihlášení k odběru více témat. Například thermostats/+/sensor/temperature/# se přihlásí k odběru všech zpráv senzoru teploty z termostatů. Konfigurace filtrů témat MQTT:

V podrobnostech o toku dat v provozním prostředí vyberte zprostředkovatele zpráv a pak pomocí pole Témata určete filtry témat MQTT pro přihlášení k odběru příchozích zpráv. Více témat MQTT můžete přidat tak, že vyberete Přidat řádek a zadáte nové téma.

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

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

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

Zástupný znak + se zde používá k výběru všech zařízení v tématech thermostats a humidifiers. Zástupný znak # se používá k výběru všech zpráv ze snímačů pod všemi dílčími tématy temperature a humidity.

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

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

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

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

Příklad s více tematickými filtry se zástupnými znaky:

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

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

Sdílená předplatná

Chcete-li použít sdílená předplatná se zdroji zprostředkovatele zpráv, můžete zadat téma sdíleného předplatného $shared/<GROUP_NAME>/<TOPIC_FILTER>ve formě .

V provozním prostředí s podrobnostmi o zdroji toku dat vyberte zprostředkovatele zpráv a pomocí pole Téma určete skupinu a téma sdíleného předplatného.

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

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ž jeden, je sdílené předplatné automaticky povoleno pro všechny toky dat, které používají zdroj zprostředkovatele zpráv. 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 používá koncový bod zprostředkovatele zpráv jako zdroj nakonfigurovaný s tématy topic1 a topic2automaticky se převedou na sdílená předplatná jako $shared/<GENERATED_GROUP_NAME>/topic1 a $shared/<GENERATED_GROUP_NAME>/topic2.

V konfiguraci můžete explicitně vytvořit téma s názvem $shared/mygroup/topic . Pokud není nutné, explicitní přidání tématu $shared se nedoporučuje, protože se předpona $shared automaticky přidá. Toky dat mohou provádět optimalizace s názvem skupiny, pokud není nastavený. $share Například není nastavená a toky dat musí pracovat jenom s názvem tématu.

Důležité

Pokud toky dat vyžadují sdílené předplatné a počet instancí je větší než jedna, je důležité používat zprostředkovatele Event Grid MQTT jako zdroj, 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 jednu při použití zprostředkovatele Event Grid MQTT jako zdroje. To znamená, že datový tok je odběratelem a přijímá zprávy z cloudu.

Témata Kafka

Pokud je zdrojem koncový bod Kafka (včetně služby Event Hubs), zadejte jednotlivá témata Kafka, k jejichž odběru se mají přihlásit příchozí zprávy. Zástupné cardy nejsou podporované, 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ý event hub v oboru názvů tématem Kafka. Pokud máte například obor názvů služby Event Hubs se dvěma event huby, thermostats a humidifiers, můžete každé event hub zadat jako Kafka téma.

Konfigurace témat Kafka:

V podrobnostech o zdroji toku dat v provozním prostředí vyberte zprostředkovatele zpráv a pak pomocí pole Téma určete filtr témat Kafka 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. Pokud chcete použít více filtrů témat, použijte Bicep nebo Kubernetes.

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

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

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

Určení zdrojového schématu

Při použití MQTT nebo Kafka jako zdroje můžete zadat schéma k zobrazení seznamu datových bodů ve webovém uživatelském rozhraní provozního prostředí. Použití schématu k deserializaci a ověření příchozích zpráv se v současné době nepodporuje.

Pokud je zdrojem prostředek, schéma se automaticky odvodí z definice prostředku.

Návod

K vygenerování schématu z ukázkového datového souboru použijte pomocníka Schema Gen.

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

V provozním prostředí s podrobnostmi o zdroji toku dat vyberte zprostředkovatele zpráv a pomocí pole Schéma zprávy určete schéma. Tlačítko Nahrát můžete použít pro nahrání souboru schématu jako první. Další informace najdete v tématu Vysvětlení schémat zpráv.

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

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

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

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

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

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

Vyžádání trvalosti disku

Trvalost disku požadavku umožňuje tokům dat zachovat stav při restartování. Když tuto funkci povolíte, graf obnoví stav zpracování, pokud se připojený zprostředkovatel restartuje. Tato funkce je užitečná ve scénářích stavového zpracování, kdy je problém se ztrátou průběžných dat. Když povolíte přetrvávání datových požadavků na disku, zprostředkovatel uloží data MQTT, například zprávy ve frontě předplatitelů, na disk. Tento přístup zajišťuje, že zdroj dat v toku dat nepřijde o data během výpadků napájení nebo restartování zprostředkovatele. Zprostředkovatel udržuje optimální výkon, protože trvalost je nakonfigurovaná na tok dat, takže tuto funkci používají jenom toky dat, které potřebují trvalost.

Graf toku dat požaduje tuto trvalost během předplatného pomocí vlastnosti uživatele MQTTv5. Tato funkce funguje pouze v těchto případech:

Tok dat používá broker MQTT nebo asset jako zdroj.
Zprostředkovatel MQTT má povolenou trvalost s dynamickým režimem trvalosti nastaveným Enabled na datový typ, jako jsou fronty odběratelů.

Tato konfigurace umožňuje klientům MQTT, jako jsou toky dat, požadovat trvalost disku pro svá předplatná pomocí vlastností uživatele MQTTv5. Podrobnosti o konfiguraci trvalosti zprostředkovatele MQTT najdete v tématu Konfigurace trvalosti zprostředkovatele MQTT.

Nastavení přijímá Enabled nebo Disabled. Disabled je výchozí.

Když vytvoříte nebo upravíte tok dat, vyberte Upravit a pak vyberte Ano vedle možnosti Požadovat trvalost dat.

Přidejte requestDiskPersistence vlastnost do konfiguračního souboru toku dat:

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

Přidejte vlastnost requestDiskPersistence do prostředku toku dat. Verze rozhraní API je 2025-10-01 nebo novější:

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

requestDiskPersistence Přidejte vlastnost do specifikace toku dat. Verze rozhraní API je connectivity.iotoperations.azure.com/v1beta1 nebo novější:

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

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 operaci transformace do konfigurace toku dat. 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:

Obohaťte: Přidejte do zdrojových dat další data na základě dané datové sady a podmínky pro shodu.
Filtr: Filtruje data na základě podmínky.
Mapování, výpočty, přejmenování nebo přidání nové vlastnosti: Přesun dat z jednoho pole do druhého pomocí volitelného převodu

Tato část představuje úvod do transformací toku dat. Podrobnější informace najdete v tématu Mapování dat pomocí toků dat, převod dat pomocí převodů toků dat a rozšiřování dat pomocí toků dat.

V prostředí provozu vyberte Tok dat>Přidat transformaci (volitelné).

Snímek obrazovky s rozhraním provozního prostředí zobrazující přidání fáze transformace do toku dat

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

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

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

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

Pokud chcete data rozšířit, nejprve přidejte referenční datovou sadu do úložiště stavů operací Azure IoT. 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ě.

Do úložiště stavů můžete načíst ukázková data pomocí CLI úložiště stavů. Názvy klíčů v úložišti stavů odpovídají datové sadě v konfiguraci toku dat.

V současné době se fáze rozšiřování nepodporuje v provozním prostředí.

K obohacení dat můžete použít builtInTransformationSettings vlastnost v konfiguraci toku dat. Vlastnost datasets slouží k určení datových sad, které se mají použít k rozšiřování.

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

Tento příklad ukazuje, jak můžete použít pole deviceId ve zdrojových datech pro spárování s polem asset 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, kde pole deviceId odpovídá thermostat1, mají pole location a manufacturer 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 na základě podmínky 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 .
Zadejte požadovaná nastavení.

Nastavení Popis

Podmínka filtru Podmínka pro filtrování dat na základě pole ve zdrojových datech.

Popis Zadejte popis podmínky filtru.

Do pole podmínky filtru zadejte @ nebo vyberte Ctrl + Mezerník a vyberte datové body z rozevíracího seznamu.

Vlastnosti metadat MQTT můžete zadat ve formátu @$metadata.user_properties.<property> nebo @$metadata.topic. Můžete také zadat záhlaví $metadata pomocí formátu @$metadata.<header>. Syntaxe $metadata je nutná pouze pro vlastnosti MQTT, které jsou součástí hlavičky zprávy. Další informace najdete v odkazech na pole.

Podmínka může používat pole ve zdrojových datech. Můžete například použít podmínku filtru, jako je @temperature > 20, k filtrování dat, která jsou menší nebo rovna 20, podle pole teploty.
Vyberte Použít.

Nastavení	Popis
Podmínka filtru	Podmínka pro filtrování dat na základě pole ve zdrojových datech.
Popis	Zadejte popis podmínky filtru.

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

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

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 rámci operačního prostředí je mapování v současné době podporováno pomocí transformací Compute, Rename a nová vlastnost.

Compute

Pomocí transformace Compute můžete aplikovat vzorec na zdrojová data. Tato operace slouží k použití vzorce na zdrojová data a uložení výsledného pole.

V části Transformace (volitelné) vyberte Výpočet>Přidat.

Zadejte požadovaná nastavení.

Nastavení	Popis
Výběr vzorce	V rozevíracím seznamu zvolte existující vzorec nebo vyberte Vlastní a zadejte vzorec ručně.
Výstup	Zadejte zobrazovaný název výstupu výsledku.
Vzorec	Zadejte vzorec, který se použije u zdrojových dat.
Popis	Zadejte popis transformace.
Poslední známá hodnota	Volitelně můžete použít poslední známou hodnotu, pokud aktuální hodnota není k dispozici.

Do pole Vzorec můžete zadat nebo upravit vzorec. Vzorec může používat pole ve zdrojových datech. Pokud chcete vybrat datové body z rozevíracího seznamu, zadejte @ nebo vyberte Ctrl + Mezerník . U předdefinovaných vzorců vyberte <dataflow> zástupný symbol a zobrazte seznam dostupných datových bodů.

Vlastnosti metadat MQTT můžete zadat ve formátu @$metadata.user_properties.<property> nebo @$metadata.topic. Můžete také zadat záhlaví $metadata pomocí formátu @$metadata.<header>. Syntaxe $metadata je nutná pouze pro vlastnosti MQTT, které jsou součástí hlavičky zprávy. Další informace najdete v odkazech na pole.

Vzorec může používat pole ve zdrojových datech. Pole ve zdrojových datech můžete například použít temperature k převodu teploty na stupně Celsia a jeho uložení do výstupního temperatureCelsius pole.

Vyberte Použít.

Přejmenovat

Datový bod můžete přejmenovat pomocí transformace Přejmenovat . Tato operace slouží k přejmenování datového bodu ve zdrojových datech na nový název. Nový název lze použít v dalších fázích toku dat.

V části Transformace (volitelné) vyberte Přejmenovat>přidat.

Zadejte požadovaná nastavení.

Nastavení	Popis
Datový bod	Vyberte datový bod z rozevíracího seznamu nebo zadejte $metadata header.
Nový název datového bodu	Zadejte nový název datového bodu.
Popis	Zadejte popis transformace.

Vyberte Použít.

Nová vlastnost

Ke zdrojovým datům můžete přidat novou vlastnost pomocí transformace nové vlastnosti . Tato operace slouží k přidání nové vlastnosti do zdrojových dat. Novou vlastnost lze použít v dalších fázích toku dat.

V části Transformace (volitelné) vyberte .
Zadejte požadovaná nastavení.

Nastavení Popis

Klíč atributu Zadejte klíč pro novou vlastnost.

Hodnota vlastnosti Zadejte hodnotu nové vlastnosti.

Popis Zadejte popis nové vlastnosti.
Vyberte Použít.

Nastavení	Popis
Klíč atributu	Zadejte klíč pro novou vlastnost.
Hodnota vlastnosti	Zadejte hodnotu nové vlastnosti.
Popis	Zadejte popis nové vlastnosti.

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:

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

K vlastnostem metadat MQTT můžete přistupovat pomocí formátu $metadata.user_properties.<property> nebo $metadata.topic. Můžete také zadat záhlaví $metadata pomocí formátu $metadata.<header>. Další informace najdete v odkazech na pole.

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řevod dat pomocí toků dat.

Odstranit

Ve výchozím nastavení jsou všechny datové body zahrnuty do výstupního schématu. Pomocí transformace Odebrat můžete z cíle odebrat libovolný datový bod.

V části Transformace (volitelné) vyberte Odebrat.
Vyberte datový bod, který chcete odebrat z výstupního schématu.
Vyberte Použít.

Pokud chcete odebrat datový bod z výstupního schématu, můžete použít builtInTransformationSettings vlastnost v konfiguraci toku dat. Vlastnost map slouží k určení datových bodů, které se mají odebrat.

{
  "operationType": "BuiltInTransformation",
  "builtInTransformationSettings": {
    "map": [
      {
        "inputs": [
          "*"
        ],
        "output": "*"
      },
      {
        "inputs": [
          "weight"
        ],
        "output": ""
      }
      {
          "inputs": [
          "weight.SourceTimestamp"
          ],
          "output": ""
      },
      {
          "inputs": [
          "weight.Value"
          ],
          "output": ""
      },
      {
          "inputs": [
          "weight.StatusCode"
          ],
          "output": ""
      },
      {
          "inputs": [
          "weight.StatusCode.Code"
          ],
          "output": ""
      },
      {
          "inputs": [
          "weight.StatusCode.Symbol"
          ],
          "output": ""
      }
    ]
  }
}

builtInTransformationSettings: {
  map: [
    {
      inputs: [
        '*'
      ]
      output: '*'
    }
    {
      inputs: [
        'weight'
      ]
      output: ''
    }
    {
      inputs: [
        'weight.SourceTimestamp'
      ]
      output: ''
    }
    {
      inputs: [
        'weight.Value'
      ]
      output: ''
    }
    {
      inputs: [
        'weight.StatusCode'
      ]
      output: ''
    }
    {
      inputs: [
        'weight.StatusCode.Code'
      ]
      output: ''
    }
    {
      inputs: [
        'weight.StatusCode.Symbol'
      ]
      output: ''
    }
  ]
}

builtInTransformationSettings:
  map:
    - type: PassThrough
      inputs:
        - "*"
      output: "*"
    - inputs:
        - weight
      output: ""
    - inputs:
        - weight.SourceTimestamp
      output: ""
    - inputs:
        - weight.Value
      output: ""
    - inputs:
        - weight.StatusCode
      output: ""
    - inputs:
        - weight.StatusCode.Code
      output: ""
    - inputs:
        - weight.StatusCode.Symbol
      output: ""

Další informace najdete v tématu Mapování dat pomocí toků dat a převod 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.

Návod

K vygenerování schématu z ukázkového datového souboru použijte pomocníka Schema Gen.

Při zkušenostech s provozem určíte schéma a formát serializace v podrobnostech koncového bodu toku dat. Koncové body, které podporují formáty serializace, jsou Microsoft Fabric OneLake, Azure Data Lake Storage Gen2, Azure Data Explorer a místní úložiště. Pokud chcete například serializovat data ve formátu Delta, musíte nahrát schéma do registru schématu a odkazovat na něj v konfiguraci cílového koncového bodu toku dat.

Snímek obrazovky využívající provozní zkušenosti k nastavení serializace cílového koncového bodu toku dat.

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

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 toku 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ílem není místní zprostředkovatel MQTT, musí se cíl použít jako zdroj. Další informace o používání místního koncového bodu zprostředkovatele MQTT 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í schéma pro serializaci. Pokud chcete použít tok dat s Microsoft Fabric OneLake, Azure Data Lake Storage, Azure Data Explorer nebo místním úložištěm, musíte zadat odkaz na schéma.

Vyberte koncový bod toku dat, který chcete použít jako cíl.

Koncové body úložiště vyžadují schéma pro serializaci. Pokud zvolíte cílový koncový bod Microsoft Fabric OneLake, Azure Data Lake Storage, Azure Data Explorer nebo místní úložiště, musíte specifikovat odkaz na schéma. Pokud například chcete data serializovat do koncového bodu Microsoft Fabric ve formátu Delta, musíte nahrát schéma do registru schématu a odkazovat na něj v konfiguraci cílového koncového bodu toku dat.
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
}

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, cílovým místem je 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í. Podporuje statická témata i dynamický překlad témat pomocí proměnných jako `${inputTopic}` a `${inputTopic.index}`. Další informace naleznete v tématu Dynamická cílová témata.
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é znaky. Pokud je koncový bod oborem názvů Event Hubs, je cílem dat jednotlivý Event Hub v rámci tohoto oboru názvů.
Azure Data Lake Storage	Kontejner	Kontejner v úložném účtu. Ne stůl.
Microsoft Fabric OneLake	Tabulka nebo složka	Odpovídá nakonfigurovaného typu cesty pro koncový bod.
Azure Data Explorer	Tabulka	Tabulka v databázi Azure Data Exploreru.
Lokální úložiště	Složka	Název složky nebo katalogu v připojení ke stálému svazku lokálního úložiště. Pokud používáte Azure Container Storage povolené službou Azure Arc Cloud Ingest Edge Volumes, musí se shodovat s `spec.path` parametrem pro subsvazek, který jste vytvořili.
OpenTelemetry	Téma	Téma OpenTelemetry, ve kterém se data odesílají. Podporují se jenom statická témata.

Chcete-li nakonfigurovat cíl 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ý bod úložiště, zobrazí se na stránce podrobností cíle výzva k zadání názvu kontejneru. Pokud je koncovým bodem toku dat koncový bod 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

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

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

U koncových bodů MQTT můžete také použít dynamické proměnné tématu. Například chcete-li směrovat zprávy z factory/1/data do processed/factory/1:

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

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

U koncových bodů MQTT můžete také použít dynamické proměnné tématu:

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

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

U koncových bodů MQTT můžete také použít dynamické proměnné tématu:

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

Dynamická cílová témata

U koncových bodů MQTT můžete pomocí dynamických proměnných tématu v dataDestination poli směrovat zprávy na základě struktury zdrojového tématu. K dispozici jsou následující proměnné:

${inputTopic} - Úplné původní téma zadávání
${inputTopic.index} - Segment vstupního tématu (index začíná na 1)

Například processed/factory/${inputTopic.2} směruje zprávy z factory/1/data do processed/factory/1. Segmenty témat jsou jedničkově indexované a počáteční a koncové lomítka se ignorují.

Pokud nelze přeložit proměnnou tématu (například ${inputTopic.5}, pokud vstupní téma obsahuje pouze tři segmenty), zpráva se zahodí a zaznamená se upozornění. Zástupné znaky (# a +) nejsou v cílových tématech povolené.

Poznámka:

Znaky $a {} jsou platné v názvech témat MQTT, takže téma factory/$inputTopic.2 je přijatelné, ale nesprávné, pokud chcete použít dynamickou proměnnou tématu.

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 MQTT tématu azure-iot-operations/data/thermostat. Transformace převede teplotu na Fahrenheita a filtruje data, ve kterých je teplota vynásobená vlhkostí menší než 1 00000. Cíl odešle data do tématu MQTT factory.

Příkaz az iot ops dataflow apply použijte k vytvoření nebo změně toku dat.

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

Parametr --config-file je cesta a název souboru JSON, který obsahuje vlastnosti prostředku.

V tomto příkladu předpokládejme konfigurační soubor s následujícím data-flow.json obsahem uloženým v domovském adresáři uživatele:

{
    "mode": "Enabled",
    "operations": [
      {
        "operationType": "Source",
        "sourceSettings": {
          "dataSources": [
            "thermostats/+/sensor/temperature/#",
            "humidifiers/+/sensor/humidity/#"
          ],
          "endpointRef": "default",
          "serializationFormat": "Json"
        }
      },
      {
        "builtInTransformationSettings": {
          "datasets": [],
          "filter": [
            {
              "expression": "$1 * $2 < 100000",
              "inputs": [
                "temperature.Value",
                "\"Tag 10\".Value"
              ],
              "type": "Filter"
            }
          ],
          "map": [
            {
              "inputs": [
                "*"
              ],
              "output": "*",
              "type": "PassThrough"
            },
            {
              "expression": "fToC($1)",
              "inputs": [
                "Temperature.Value"
              ],
              "output": "TemperatureF",
              "type": "Compute"
            },
            {
              "inputs": [
                "@\"Tag 10\".Value"
              ],
              "output": "Humidity",
              "type": "Rename"
            }
          ],
          "serializationFormat": "Json"
        },
        "operationType": "BuiltInTransformation"
      },
      {
        "destinationSettings": {
          "dataDestination": "factory",
          "endpointRef": "default"
        },
        "operationType": "Destination"
      }
    ]
}

Tady je ukázkový příkaz pro vytvoření nebo aktualizaci toku dat pomocí výchozího profilu toku dat:

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

Tady je další příklad použití dynamického překladu témat ke směrování zpráv z různých termostatů do témat specifických pro zařízení:

{
    "mode": "Enabled",
    "operations": [
      {
        "operationType": "Source",
        "sourceSettings": {
          "dataSources": [
            "thermostats/+/sensor/temperature"
          ],
          "endpointRef": "default",
          "serializationFormat": "Json"
        }
      },
      {
        "destinationSettings": {
          "dataDestination": "processed/device/${inputTopic.2}/temperature",
          "endpointRef": "default"
        },
        "operationType": "Destination"
      }
    ]
}

Tato konfigurace zpracovává zprávy z thermostats/device1/sensor/temperature a odesílá je do processed/device/device1/temperature.

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

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

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

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

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

resource dataflow 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflows@2024-11-01' = {
  // Reference to the parent data flow profile, the default profile in this case
  // Same usage as profileRef in Kubernetes YAML
  parent: defaultDataflowProfile
  name: dataflowName
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    mode: 'Enabled'
    operations: [
      {
        operationType: 'Source'
        sourceSettings: {
          // Use the default MQTT endpoint as the source
          endpointRef: defaultDataflowEndpoint.name
          // Filter the data from the MQTT topic azure-iot-operations/data/thermostat
          dataSources: [
            'azure-iot-operations/data/thermostat'
          ]
        }
      }
      // Transformation optional
      {
        operationType: 'BuiltInTransformation'
        builtInTransformationSettings: {
          // Filter the data where temperature * "Tag 10" < 100000
          filter: [
            {
              inputs: [
                'temperature.Value'
                '"Tag 10".Value'
              ]
              expression: '$1 * $2 < 100000'
            }
          ]
          map: [
            // Passthrough all values by default
            {
              inputs: [
                '*'
              ]
              output: '*'
            }
            // Convert temperature to Fahrenheit and output it to TemperatureF
            {
              inputs: [
                'temperature.Value'
              ]
              output: 'TemperatureF'
              expression: 'cToF($1)'
            }
          // Extract the "Tag 10" value and output it to Humidity
            {
              inputs: [
                '"Tag 10".Value'
              ]
              output: 'Humidity'
            }
          ]
        }
      }
      {
        operationType: 'Destination'
        destinationSettings: {
          // Use the default MQTT endpoint as the destination
          endpointRef: defaultDataflowEndpoint.name
          // Send the data to the MQTT topic factory
          dataDestination: 'factory'
        }
      }
    ]
  }
}

apiVersion: connectivity.iotoperations.azure.com/v1
kind: Dataflow
metadata:
  name: my-dataflow
  namespace: azure-iot-operations
spec:
  # Reference to the default data flow profile
  profileRef: default
  mode: Enabled
  operations:
    - operationType: Source
      sourceSettings:
        # Use the default MQTT endpoint as the source
        endpointRef: default
        # Filter the data from the MQTT topic azure-iot-operations/data/thermostat
        dataSources:
          - azure-iot-operations/data/thermostat
    # Transformation optional
    - operationType: builtInTransformation
      builtInTransformationSettings:
        # Filter the data where temperature * "Tag 10" < 100000
        filter:
          - inputs:
              - 'temperature.Value'
              - '"Tag 10".Value'
            expression: '$1 * $2 < 100000'
        map:
          # Passthrough all values by default
          - inputs:
              - '*'
            output: '*'
          # Convert temperature to Fahrenheit and output it to TemperatureF
          - inputs:
              - temperature.Value
            output: TemperatureF
            expression: cToF($1)
          # Extract the "Tag 10" value and output it to Humidity
          - inputs:
              - '"Tag 10".Value'
            output: 'Humidity'
    - operationType: Destination
      destinationSettings:
        # Use the default MQTT endpoint as the destination
        endpointRef: default
        # Send the data to the MQTT topic factory
        dataDestination: factory

Další příklady konfigurací toku dat najdete v tématu Azure REST API – Tok dat a Rychlý průvodce 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.

Konfigurace exportu 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 rozhraním provozního prostředí zobrazující možnost exportu pro nakonfigurovaný tok dat

K exportu toku dat použijte příkaz az iot ops dataflow show .

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

Tady je ukázkový příkaz pro export toku dat pojmenovaného data-flow do souboru JSON s názvem data-flow.json:

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

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

Správná konfigurace toku dat

Pokud chcete zajistit, aby tok dat fungoval očekávaným způsobem, ověřte následující:

Jako zdroj nebo cíl se musí použít výchozí koncový bod toku dat MQTT.
Profil toku dat existuje a odkazuje se na ho v konfiguraci toku dat.
Zdrojem je koncový bod MQTT, koncový bod Kafka nebo aktivum. 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í Event Hubs jako zdroje je každý Event Hub v rámci oboru názvů samostatným Kafka tématem a musí být specifikován 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 symbolů.
Při použití koncových bodů typu úložiště jako cíle se specifikuje schéma.
Při použití dynamických cílových témat pro koncové body MQTT se ujistěte, že proměnné tématu odkazují na platné segmenty.

Další kroky

Váš názor

Byla tato stránka užitečná?

Last updated on 2025-10-02

Sdílet prostřednictvím

Vytváření toků dat v operacích Azure IoT

Požadavky

Profil toku dat

Koncové body toku dat

Začínáme

Zdroj

Možnost 1: Použití výchozího koncového bodu zprostředkovatele zpráv jako zdroje

Možnost 2: Použití materiálu jako zdroje

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

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

Témata MQTT

Sdílená předplatná

Témata Kafka

Určení zdrojového schématu

Vyžádání trvalosti disku

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

Compute

Přejmenovat

Nová vlastnost

Odstranit

Serializace dat podle schématu

Cíl

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

Dynamická cílová témata

Příklad

Ověření, že tok dat funguje

Konfigurace exportu toku dat

Správná konfigurace toku dat

Další kroky

Váš názor

Další materiály