Konfigurace toků dat v operacích Azure IoT

Článek
12/12/2024

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 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 operací Azure IoT 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

Pokud pro toky dat nepotřebujete různá nastavení škálování, použijte výchozí profil toku dat poskytovaný operacemi Azure IoT. 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-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
        }
      }
    ]
  }
}

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

Možnost 2: 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.

Možnost 3: Použití vlastního koncového bodu MQTT nebo toku dat Kafka 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ž jeden, sdílené předplatné se automaticky povolí 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.

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

Důležité

Toky dat vyžadující sdílené předplatné, pokud je počet instancí větší než jeden, 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 jednu 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í zdrojového schématu

Při použití MQTT nebo Kafka jako zdroje můžete zadat schéma pro zobrazení seznamu datových bodů na portálu provozního prostředí. Všimněte si, že 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.

Tip

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

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

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řidejte do zdrojových dat další data zadanou datovou sadu a podmínku, která se mají shodovat.
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 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

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

Ukázková data můžete načíst do úložiště stavů pomocí rozhraní příkazového řádku ú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í.

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>.
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 $metadata záhlaví 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 @temperature > 20 je filtrování dat menší než nebo rovno 20 na základě 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:

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í vlastností Compute, Rename a New.

Compute

Pomocí transformace Compute můžete použít 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 Přidat výpočetní prostředky>.

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 .

Vlastnosti metadat MQTT můžete zadat ve formátu @$metadata.user_properties.<property> nebo @$metadata.topic. Můžete také zadat $metadata záhlaví 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 následujících fázích toku dat.

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

Zadejte požadovaná nastavení.

Nastavení	Popis
Datapoint	V rozevíracím seznamu vyberte datový bod nebo zadejte $metadata záhlaví.
Nový název datového bodu	Zadejte nový název datového bodu.
Popis	Zadejte popis transformace.

Pokud chcete vybrat datové body z rozevíracího seznamu, zadejte @ nebo vyberte Ctrl + Mezerník .

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 následujících fázích toku dat.

V části Transformace (volitelné) vyberte Přidat novou vlastnost>.
Zadejte požadovaná nastavení.

Nastavení Popis

Klíč vlastnosti Zadejte klíč nové vlastnosti.

Hodnota vlastnosti Zadejte hodnotu nové vlastnosti.

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

Nastavení	Popis
Klíč vlastnosti	Zadejte klíč nové vlastnosti.
Hodnota vlastnosti	Zadejte hodnotu nové vlastnosti.
Popis	Zadejte popis nové vlastnosti.

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

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.

Tip

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

V případě provozního prostředí zadáte v podrobnostech koncového bodu toku dat formát schématu a serializace. Koncové body, které podporují formáty serializace, jsou Microsoft Fabric OneLake, Azure Data Lake Storage Gen2 a Azure Data Explorer. 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 s provozním prostředím 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>'

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í schéma pro serializaci. Pokud chcete použít tok dat s Microsoft Fabric OneLake, Azure Data Lake Storage, Azure Data Explorerem 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 zadat 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

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 filtruje 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-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

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

Možnost 1: Použití výchozího koncového bodu MQTT jako zdroje

Možnost 2: Použití prostředku jako zdroje

Možnost 3: Použití vlastního koncového bodu MQTT nebo toku dat 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

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

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