Konfigurowanie przepływów danych w operacjach usługi Azure IoT

Ważne

Ta strona zawiera instrukcje dotyczące zarządzania składnikami operacji usługi Azure IoT przy użyciu manifestów wdrażania platformy Kubernetes, które są w wersji zapoznawczej. Ta funkcja jest udostępniana z kilkoma ograniczeniami i nie powinna być używana w przypadku obciążeń produkcyjnych.

Zobacz Dodatkowe warunki użytkowania wersji zapoznawczych platformy Microsoft Azure, aby zapoznać się z postanowieniami prawnymi dotyczącymi funkcji platformy Azure, które są w wersji beta lub wersji zapoznawczej albo w inny sposób nie zostały jeszcze wydane jako ogólnie dostępne.

Przepływ danych to ścieżka pobierana od źródła do miejsca docelowego z opcjonalnymi przekształceniami. Przepływ danych można skonfigurować, tworząc zasób niestandardowy przepływu danych lub korzystając z portalu programu Azure IoT Operations Studio. Przepływ danych składa się z trzech części: źródła, transformacji i miejsca docelowego.

Aby zdefiniować źródło i miejsce docelowe, należy skonfigurować punkty końcowe przepływu danych. Transformacja jest opcjonalna i może obejmować operacje, takie jak wzbogacanie danych, filtrowanie danych i mapowanie danych na inne pole.

Ważne

Każdy przepływ danych musi mieć lokalny punkt końcowy brokera MQTT operacji usługi Azure IoT jako źródło lub miejsce docelowe.

Aby utworzyć przepływ danych, możesz użyć środowiska operacji w operacjach usługi Azure IoT. Środowisko operacji udostępnia wizualny interfejs do konfigurowania przepływu danych. Możesz również użyć narzędzia Bicep, aby utworzyć przepływ danych przy użyciu pliku szablonu Bicep lub utworzyć przepływ danych przy użyciu pliku YAML.

Kontynuuj czytanie, aby dowiedzieć się, jak skonfigurować źródło, transformację i miejsce docelowe.

Wymagania wstępne

Przepływy danych można wdrażać natychmiast po wystąpieniu operacji usługi Azure IoT przy użyciu domyślnego profilu przepływu danych i punktu końcowego. Możesz jednak skonfigurować profile i punkty końcowe przepływu danych, aby dostosować przepływ danych.

Profil przepływu danych

Jeśli nie potrzebujesz różnych ustawień skalowania dla przepływów danych, użyj domyślnego profilu przepływu danych udostępnianego przez operacje usługi Azure IoT. Aby dowiedzieć się, jak skonfigurować profil przepływu danych, zobacz Konfigurowanie profilów przepływu danych.

Punkty końcowe przepływu danych

Punkty końcowe przepływu danych są wymagane do skonfigurowania źródła i miejsca docelowego dla przepływu danych. Aby szybko rozpocząć pracę, możesz użyć domyślnego punktu końcowego przepływu danych dla lokalnego brokera MQTT. Można również tworzyć inne typy punktów końcowych przepływu danych, takich jak Kafka, Event Hubs lub Azure Data Lake Storage. Aby dowiedzieć się, jak skonfigurować każdy typ punktu końcowego przepływu danych, zobacz Konfigurowanie punktów końcowych przepływu danych.

Rozpocznij

Po spełnieniu wymagań wstępnych możesz rozpocząć tworzenie przepływu danych.

Portal

Aby utworzyć przepływ danych w środowisku operacji, wybierz pozycję Przepływ>danych Utwórz przepływ danych. Następnie zostanie wyświetlona strona, na której można skonfigurować źródło, przekształcenie i miejsce docelowe dla przepływu danych.

Bicep

Utwórz plik Bicep .bicep , aby rozpocząć tworzenie przepływu danych. W tym przykładzie pokazano strukturę przepływu danych zawierającego konfigurację źródłową, transformację i lokalizację docelową.

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

Kubernetes (wersja zapoznawcza)

Utwórz plik manifestu .yaml kubernetes, aby rozpocząć tworzenie przepływu danych. W tym przykładzie pokazano strukturę przepływu danych zawierającego konfigurację źródłową, transformację i lokalizację docelową.

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

Zapoznaj się z poniższymi sekcjami, aby dowiedzieć się, jak skonfigurować typy operacji przepływu danych.

Źródło

Aby skonfigurować źródło przepływu danych, określ odwołanie do punktu końcowego i listę źródeł danych dla punktu końcowego. Wybierz jedną z następujących opcji jako źródło przepływu danych.

Jeśli domyślny punkt końcowy nie jest używany jako źródło, musi być używany jako miejsce docelowe. Aby dowiedzieć się więcej, zobacz Przepływy danych muszą używać lokalnego punktu końcowego brokera MQTT.

Opcja 1. Użyj domyślnego punktu końcowego MQTT jako źródła

Portal

1. W obszarze Szczegóły źródła wybierz pozycję MQTT.

   

2. Wprowadź następujące ustawienia dla źródła MQTT:

   Ustawienie	opis
   Temat MQTT	Filtr tematu MQTT, aby subskrybować komunikaty przychodzące. Zobacz Konfigurowanie tematów MQTT lub Kafka.
   Schemat komunikatu	Schemat używany do deserializacji przychodzących komunikatów. Zobacz Określanie schematu do deserializacji danych.

3. Wybierz Zastosuj.

Bicep

Punkt końcowy MQTT jest skonfigurowany w pliku szablonu Bicep. Na przykład następujący punkt końcowy jest źródłem przepływu danych.

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

dataSources W tym miejscu można określić wiele tematów MQTT lub Kafka bez konieczności modyfikowania konfiguracji punktu końcowego. Ta elastyczność oznacza, że ten sam punkt końcowy może być ponownie używany w wielu przepływach danych, nawet jeśli tematy się różnią. Aby dowiedzieć się więcej, zobacz Konfigurowanie źródeł danych.

Kubernetes (wersja zapoznawcza)

Aby na przykład skonfigurować źródło przy użyciu punktu końcowego MQTT i dwóch filtrów tematu MQTT, użyj następującej konfiguracji:

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

Ponieważ dataSources umożliwia określenie tematów MQTT lub Kafka bez modyfikowania konfiguracji punktu końcowego, można ponownie użyć punktu końcowego dla wielu przepływów danych, nawet jeśli tematy są inne. Aby dowiedzieć się więcej, zobacz Konfigurowanie źródeł danych.

Opcja 2. Użyj elementu zawartości jako źródła

Portal

Zasób można użyć jako źródła przepływu danych. Używanie elementu zawartości jako źródła jest dostępne tylko w środowisku operacji.

1. W obszarze Szczegóły źródła wybierz pozycję Zasób.

2. Wybierz zasób, którego chcesz użyć jako źródłowego punktu końcowego.

3. Wybierz pozycję Kontynuuj.

   Zostanie wyświetlona lista punktów danych dla wybranego zasobu.

   

4. Wybierz pozycję Zastosuj , aby użyć elementu zawartości jako źródłowego punktu końcowego.

Bicep

Konfigurowanie zasobu jako źródła jest dostępne tylko w środowisku operacji.

Kubernetes (wersja zapoznawcza)

Konfigurowanie zasobu jako źródła jest dostępne tylko w środowisku operacji.

W przypadku używania zasobu jako źródła definicja zasobu służy do wnioskowania schematu przepływu danych. Definicja zasobu zawiera schemat punktów danych zasobu. Aby dowiedzieć się więcej, zobacz Zdalne zarządzanie konfiguracjami zasobów.

Po skonfigurowaniu dane z zasobu dotarły do przepływu danych za pośrednictwem lokalnego brokera MQTT. Dlatego w przypadku używania elementu zawartości jako źródła przepływ danych używa domyślnego punktu końcowego lokalnego brokera MQTT jako źródła w rzeczywistości.

Opcja 3. Użyj niestandardowego punktu końcowego przepływu danych MQTT lub Kafka jako źródła

Jeśli utworzono niestandardowy punkt końcowy przepływu danych MQTT lub Kafka (na przykład do użycia z usługą Event Grid lub Event Hubs), możesz użyć go jako źródła przepływu danych. Pamiętaj, że punkty końcowe typu magazynu, takie jak Data Lake lub Fabric OneLake, nie mogą być używane jako źródło.

Aby skonfigurować, użyj rozwiązania Kubernetes YAML lub Bicep. Zastąp wartości symboli zastępczych niestandardową nazwą punktu końcowego i tematami.

Portal

Używanie niestandardowego punktu końcowego MQTT lub Kafka jako źródła nie jest obecnie obsługiwane w środowisku operacji.

Bicep

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

Kubernetes (wersja zapoznawcza)

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

Konfigurowanie źródeł danych (tematy MQTT lub Kafka)

W źródle można określić wiele tematów MQTT lub Kafka bez konieczności modyfikowania konfiguracji punktu końcowego przepływu danych. Ta elastyczność oznacza, że ten sam punkt końcowy może być ponownie używany w wielu przepływach danych, nawet jeśli tematy się różnią. Aby uzyskać więcej informacji, zobacz Ponowne używanie punktów końcowych przepływu danych.

Tematy MQTT

Jeśli źródłem jest punkt końcowy MQTT (uwzględniona usługa Event Grid), możesz użyć filtru tematu MQTT, aby subskrybować komunikaty przychodzące. Filtr tematu może zawierać symbole wieloznaczne, aby subskrybować wiele tematów. Na przykład thermostats/+/telemetry/temperature/# subskrybuje wszystkie komunikaty telemetryczne dotyczące temperatury z termostatów. Aby skonfigurować filtry tematów MQTT:

Portal

W obszarze Szczegóły źródła przepływu danych środowiska operacji wybierz pozycję MQTT, a następnie użyj pola tematu MQTT, aby określić filtr tematu MQTT, aby zasubskrybować komunikaty przychodzące.

Uwaga

W środowisku operacji można określić tylko jeden filtr tematu MQTT. Aby użyć wielu filtrów tematów MQTT, użyj Bicep lub Kubernetes.

Bicep

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

Przykład z wieloma filtrami tematów MQTT z symbolami wieloznacznymi:

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

W tym miejscu symbol wieloznaczny + służy do wybierania wszystkich urządzeń w obszarze thermostats tematów i humidifiers . Symbol # wieloznaczny służy do wybierania wszystkich komunikatów telemetrycznych we wszystkich podtopiach tematów temperature i humidity .

Kubernetes (wersja zapoznawcza)

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

Przykład z wieloma filtrami tematów MQTT z symbolami wieloznacznymi:

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

W tym miejscu symbol wieloznaczny + służy do wybierania wszystkich urządzeń w obszarze thermostats tematów i humidifiers . Symbol # wieloznaczny służy do wybierania wszystkich komunikatów telemetrycznych we wszystkich podtopiach tematów temperature i humidity .

Subskrypcje udostępnione

Aby użyć subskrypcji udostępnionych ze źródłami MQTT, możesz określić temat subskrypcji udostępnionej $shared/<GROUP_NAME>/<TOPIC_FILTER>w postaci .

Portal

W obszarze Szczegóły źródła przepływu danych środowiska operacji wybierz pozycję MQTT i użyj pola tematu MQTT, aby określić udostępnioną grupę subskrypcji i temat.

Bicep

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

Kubernetes (wersja zapoznawcza)

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

Jeśli liczba wystąpień w profilu przepływu danych jest większa niż jedna, subskrypcja udostępniona jest automatycznie włączona dla wszystkich przepływów danych korzystających ze źródła MQTT. W takim przypadku $shared prefiks jest dodawany, a nazwa udostępnionej grupy subskrypcji jest generowana automatycznie. Jeśli na przykład masz profil przepływu danych z liczbą wystąpień wynoszącą 3, a przepływ danych używa punktu końcowego MQTT jako źródła skonfigurowanego z tematami topic1 i topic2, są one automatycznie konwertowane na subskrypcje udostępnione jako $shared/<GENERATED_GROUP_NAME>/topic1 i $shared/<GENERATED_GROUP_NAME>/topic2.

Możesz jawnie utworzyć temat o nazwie $shared/mygroup/topic w konfiguracji. Jednak jawne dodanie tematu $shared nie jest zalecane, ponieważ $shared prefiks jest automatycznie dodawany w razie potrzeby. Przepływy danych mogą dokonać optymalizacji przy użyciu nazwy grupy, jeśli nie jest ustawiona. Na przykład nie jest ustawiona, $share a przepływy danych muszą działać tylko nad nazwą tematu.

Ważne

Przepływy danych wymagające subskrypcji udostępnionej, gdy liczba wystąpień jest większa niż jedna, jest ważna w przypadku korzystania z brokera MQTT usługi Event Grid jako źródła, ponieważ nie obsługuje subskrypcji udostępnionych. Aby uniknąć brakujących komunikatów, ustaw liczbę wystąpień profilu przepływu danych na jedną podczas korzystania z brokera MQTT usługi Event Grid jako źródła. Dzieje się tak, gdy przepływ danych jest subskrybentem i odbiera komunikaty z chmury.

Tematy dotyczące platformy Kafka

Gdy źródłem jest punkt końcowy platformy Kafka (dołączone do usługi Event Hubs), określ poszczególne tematy platformy Kafka, które mają być subskrybowane dla komunikatów przychodzących. Symbole wieloznaczne nie są obsługiwane, dlatego należy statycznie określić każdy temat.

Uwaga

W przypadku korzystania z usługi Event Hubs za pośrednictwem punktu końcowego platformy Kafka każde pojedyncze centrum zdarzeń w przestrzeni nazw jest tematem platformy Kafka. Jeśli na przykład masz przestrzeń nazw usługi Event Hubs z dwoma centrami zdarzeń i thermostats humidifiers, możesz określić każde centrum zdarzeń jako temat platformy Kafka.

Aby skonfigurować tematy platformy Kafka:

Portal

Używanie punktu końcowego platformy Kafka jako źródła nie jest obecnie obsługiwane w środowisku operacji.

Bicep

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

Kubernetes (wersja zapoznawcza)

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

Określanie schematu źródłowego

W przypadku używania MQTT lub Kafka jako źródła można określić schemat umożliwiający wyświetlenie listy punktów danych w portalu środowiska operacji. Należy pamiętać, że używanie schematu do deserializacji i weryfikowania przychodzących komunikatów nie jest obecnie obsługiwane.

Jeśli źródło jest zasobem, schemat jest automatycznie wnioskowany z definicji zasobu.

Napiwek

Aby wygenerować schemat na podstawie przykładowego pliku danych, użyj pomocnika generacji schematu.

Aby skonfigurować schemat używany do deserializacji przychodzących komunikatów ze źródła:

Portal

W obszarze Szczegóły źródła przepływu danych środowiska operacji wybierz pozycję MQTT i użyj pola Schemat komunikatu, aby określić schemat. Możesz użyć przycisku Przekaż, aby najpierw przekazać plik schematu. Aby dowiedzieć się więcej, zobacz Omówienie schematów komunikatów.

Bicep

Po użyciu rejestru schematów do przechowywania schematu można odwoływać się do niego w konfiguracji przepływu danych.

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

Kubernetes (wersja zapoznawcza)

Po użyciu rejestru schematów do przechowywania schematu można odwoływać się do niego w konfiguracji przepływu danych.

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

Aby dowiedzieć się więcej, zobacz Omówienie schematów komunikatów.

Przekształcenie

Operacja przekształcania umożliwia przekształcenie danych ze źródła przed wysłaniem ich do miejsca docelowego. Przekształcenia są opcjonalne. Jeśli nie musisz wprowadzać zmian w danych, nie uwzględniaj operacji przekształcania w konfiguracji przepływu danych. Wiele przekształceń jest ze sobą łańcuchowych w etapach niezależnie od kolejności, w jakiej są określone w konfiguracji. Kolejność etapów jest zawsze:

1. Wzbogacanie: dodaj dodatkowe dane do danych źródłowych, biorąc pod uwagę zestaw danych i warunek, które mają być zgodne.
2. Filtr: filtruj dane na podstawie warunku.
3. Mapowanie, obliczanie, zmienianie nazwy lub dodawanie nowej właściwości: przenoszenie danych z jednego pola do innego z opcjonalną konwersją.

Ta sekcja stanowi wprowadzenie do przekształceń przepływu danych. Aby uzyskać bardziej szczegółowe informacje, zobacz Mapuj dane przy użyciu przepływów danych, Konwertuj dane przy użyciu konwersji przepływu danych i Wzbogacaj dane przy użyciu przepływów danych.

Portal

W środowisku operacji wybierz pozycję Przepływ>danych Dodaj przekształcenie (opcjonalnie)..

Bicep

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

Kubernetes (wersja zapoznawcza)

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

Wzbogacanie: dodawanie danych referencyjnych

Aby wzbogacić dane, najpierw dodaj zestaw danych referencyjnych w magazynie stanu operacji usługi Azure IoT. Zestaw danych służy do dodawania dodatkowych danych do danych źródłowych na podstawie warunku. Warunek jest określany jako pole w danych źródłowych, które pasuje do pola w zestawie danych.

Przykładowe dane można załadować do magazynu stanów przy użyciu interfejsu wiersza polecenia magazynu stanów. Nazwy kluczy w magazynie stanów odpowiadają zestawowi danych w konfiguracji przepływu danych.

Portal

Obecnie etap Wzbogacanie nie jest obsługiwany w środowisku operacji.

Bicep

W tym przykładzie pokazano, jak można użyć deviceId pola w danych źródłowych, aby dopasować asset je do pola w zestawie danych:

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

Kubernetes (wersja zapoznawcza)

Na przykład możesz użyć deviceId pola w danych źródłowych, aby dopasować asset je do pola w zestawie danych:

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

Jeśli zestaw danych zawiera rekord z polem asset , podobnie jak:

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

Dane ze źródła z pasującym polem deviceId mają location pola i manufacturer dostępne w etapach filtrowania thermostat1 i mapy.

Aby uzyskać więcej informacji na temat składni warunku, zobacz Wzbogacanie danych przy użyciu przepływów danych i Konwertowanie danych przy użyciu przepływów danych.

Filtr: filtrowanie danych na podstawie warunku

Aby filtrować dane według warunku, możesz użyć etapu filter . Warunek jest określony jako pole w danych źródłowych, które pasuje do wartości.

Portal

1. W obszarze Przekształć (opcjonalnie) wybierz pozycję Filtr>Dodaj.

   

2. Wprowadź wymagane ustawienia.

   Ustawienie	opis
   Warunek filtru	Warunek filtrowania danych na podstawie pola w danych źródłowych.
   opis	Podaj opis warunku filtru.

   W polu warunek filtru wprowadź @ lub wybierz Ctrl + Spacja , aby wybrać punkty danych z listy rozwijanej.

   Właściwości metadanych MQTT można wprowadzić przy użyciu formatu @$metadata.user_properties.<property> lub @$metadata.topic. Możesz również wprowadzić nagłówki $metadata przy użyciu formatu @$metadata.<header>. Składnia jest wymagana $metadata tylko dla właściwości MQTT, które są częścią nagłówka komunikatu. Aby uzyskać więcej informacji, zobacz odwołania do pól.

   Warunek może używać pól w danych źródłowych. Można na przykład użyć warunku filtru, takiego jak @temperature > 20 filtrowanie danych mniejszych lub równych 20 na podstawie pola temperatury.

3. Wybierz Zastosuj.

Bicep

Na przykład możesz użyć temperature pola w danych źródłowych, aby filtrować dane:

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

temperature Jeśli pole jest większe niż 20, dane są przekazywane do następnego etapu. temperature Jeśli pole jest mniejsze lub równe 20, dane są filtrowane.

Kubernetes (wersja zapoznawcza)

Na przykład możesz użyć temperature pola w danych źródłowych, aby filtrować dane:

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

temperature Jeśli pole jest większe niż 20, dane są przekazywane do następnego etapu. temperature Jeśli pole jest mniejsze lub równe 20, dane są filtrowane.

Mapa: Przenoszenie danych z jednego pola do innego

Aby zamapować dane na inne pole z opcjonalną konwersją, możesz użyć map operacji . Konwersja jest określana jako formuła, która używa pól w danych źródłowych.

Portal

W środowisku operacji mapowanie jest obecnie obsługiwane przy użyciu przekształceń właściwości Compute, Rename i New .

Compute

Możesz użyć przekształcenia Obliczenia , aby zastosować formułę do danych źródłowych. Ta operacja służy do stosowania formuły do danych źródłowych i przechowywania pola wyników.

1. W obszarze Przekształć (opcjonalnie) wybierz pozycję Dodaj obliczenia>.

   

2. Wprowadź wymagane ustawienia.

   Ustawienie	opis
   Wybieranie formuły	Wybierz istniejącą formułę z listy rozwijanej lub wybierz pozycję Niestandardowe , aby ręcznie wprowadzić formułę.
   Wyjście	Określ nazwę wyświetlaną danych wyjściowych dla wyniku.
   Formuła	Wprowadź formułę, która ma zostać zastosowana do danych źródłowych.
   opis	Podaj opis przekształcenia.
   Ostatnia znana wartość	Opcjonalnie użyj ostatniej znanej wartości, jeśli bieżąca wartość jest niedostępna.

   Formułę można wprowadzić lub edytować w polu Formuła . Formuła może używać pól w danych źródłowych. Wprowadź @ lub wybierz Ctrl + Spacja , aby wybrać punkty danych z listy rozwijanej.

   Właściwości metadanych MQTT można wprowadzić przy użyciu formatu @$metadata.user_properties.<property> lub @$metadata.topic. Możesz również wprowadzić nagłówki $metadata przy użyciu formatu @$metadata.<header>. Składnia jest wymagana $metadata tylko dla właściwości MQTT, które są częścią nagłówka komunikatu. Aby uzyskać więcej informacji, zobacz odwołania do pól.

   Formuła może używać pól w danych źródłowych. Możesz na przykład użyć temperature pola w danych źródłowych, aby przekonwertować temperaturę na stopnie Celsjusza i zapisać je w polu wyjściowym temperatureCelsius .

3. Wybierz Zastosuj.

Zmień nazwę

Nazwę punktu danych można zmienić przy użyciu przekształcenia Zmień nazwę . Ta operacja służy do zmieniania nazwy punktu danych w danych źródłowych na nową nazwę. Nowa nazwa może być używana w kolejnych etapach przepływu danych.

1. W obszarze Przekształć (opcjonalnie)wybierz pozycję Zmień nazwę>dodaj.

   

2. Wprowadź wymagane ustawienia.

   Ustawienie	opis
   Datapoint	Wybierz punkt danych z listy rozwijanej lub wprowadź nagłówek $metadata.
   Nowa nazwa punktu danych	Wprowadź nową nazwę punktu danych.
   opis	Podaj opis przekształcenia.

   Wprowadź @ lub wybierz Ctrl + Spacja , aby wybrać punkty danych z listy rozwijanej.

   Właściwości metadanych MQTT można wprowadzić przy użyciu formatu @$metadata.user_properties.<property> lub @$metadata.topic. Możesz również wprowadzić nagłówki $metadata przy użyciu formatu @$metadata.<header>. Składnia jest wymagana $metadata tylko dla właściwości MQTT, które są częścią nagłówka komunikatu. Aby uzyskać więcej informacji, zobacz odwołania do pól.

3. Wybierz Zastosuj.

Nowa właściwość

Możesz dodać nową właściwość do danych źródłowych przy użyciu przekształcenia nowej właściwości . Ta operacja służy do dodawania nowej właściwości do danych źródłowych. Nowa właściwość może być używana w kolejnych etapach przepływu danych.

1. W obszarze Przekształć (opcjonalnie)wybierz pozycję Nowa właściwość>Dodaj.

   

2. Wprowadź wymagane ustawienia.

   Ustawienie	opis
   Klucz właściwości	Wprowadź klucz nowej właściwości.
   Wartości właściwości	Wprowadź wartość nowej właściwości.
   opis	Podaj opis nowej właściwości.

3. Wybierz Zastosuj.

Bicep

Dostęp do właściwości metadanych MQTT można uzyskać przy użyciu formatu $metadata.user_properties.<property> lub $metadata.topic. Możesz również wprowadzić nagłówki $metadata przy użyciu formatu $metadata.<header>. Aby uzyskać więcej informacji, zobacz odwołania do pól.

Możesz na przykład użyć temperature pola w danych źródłowych, aby przekonwertować temperaturę na stopnie Celsjusza i zapisać je w temperatureCelsius polu. Możesz również wzbogacić dane źródłowe za pomocą location pola z zestawu danych kontekstowego:

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

Kubernetes (wersja zapoznawcza)

Dostęp do właściwości metadanych MQTT można uzyskać przy użyciu formatu $metadata.user_properties.<property> lub $metadata.topic. Możesz również wprowadzić nagłówki $metadata przy użyciu formatu $metadata.<header>. Aby uzyskać więcej informacji, zobacz odwołania do pól.

Możesz na przykład użyć temperature pola w danych źródłowych, aby przekonwertować temperaturę na stopnie Celsjusza i zapisać je w temperatureCelsius polu. Możesz również wzbogacić dane źródłowe za pomocą location pola z zestawu danych kontekstowego:

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

Aby dowiedzieć się więcej, zobacz Mapuj dane przy użyciu przepływów danych i Konwertuj dane przy użyciu przepływów danych.

Serializowanie danych zgodnie ze schematem

Jeśli chcesz serializować dane przed wysłaniem ich do miejsca docelowego, musisz określić format schematu i serializacji. W przeciwnym razie dane są serializowane w formacie JSON z typami wywnioskowanymi. Punkty końcowe magazynu, takie jak Microsoft Fabric lub Azure Data Lake, wymagają schematu w celu zapewnienia spójności danych. Obsługiwane formaty serializacji to Parquet i Delta.

Napiwek

Aby wygenerować schemat na podstawie przykładowego pliku danych, użyj pomocnika generacji schematu.

Portal

W przypadku środowiska operacji należy określić format schematu i serializacji w szczegółach punktu końcowego przepływu danych. Punkty końcowe, które obsługują formaty serializacji, to Microsoft Fabric OneLake, Azure Data Lake Storage Gen 2 i Azure Data Explorer. Na przykład aby serializować dane w formacie delta, należy przekazać schemat do rejestru schematów i odwołać się do niego w konfiguracji docelowego punktu końcowego przepływu danych.

Bicep

Po przekazaniu schematu do rejestru schematów można odwoływać się do niego w konfiguracji przepływu danych.

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

Kubernetes (wersja zapoznawcza)

Po przekazaniu schematu do rejestru schematów można odwoływać się do niego w konfiguracji przepływu danych.

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

Aby uzyskać więcej informacji na temat rejestru schematów, zobacz Omówienie schematów komunikatów.

Element docelowy

Aby skonfigurować miejsce docelowe dla przepływu danych, określ odwołanie do punktu końcowego i miejsce docelowe danych. Możesz określić listę miejsc docelowych danych dla punktu końcowego.

Aby wysłać dane do miejsca docelowego innego niż lokalny broker MQTT, utwórz punkt końcowy przepływu danych. Aby dowiedzieć się, jak to zrobić, zobacz Konfigurowanie punktów końcowych przepływu danych. Jeśli miejsce docelowe nie jest lokalnym brokerem MQTT, musi być używane jako źródło. Aby dowiedzieć się więcej, zobacz Przepływy danych muszą używać lokalnego punktu końcowego brokera MQTT.

Ważne

Punkty końcowe magazynu wymagają schematu serializacji. Aby używać przepływu danych z usługą Microsoft Fabric OneLake, Azure Data Lake Storage, Azure Data Explorer lub Local Storage, należy określić odwołanie do schematu.

Portal

1. Wybierz punkt końcowy przepływu danych, który ma być używany jako miejsce docelowe.

   

   Punkty końcowe magazynu wymagają schematu serializacji. Jeśli wybierzesz punkt końcowy docelowy usługi Microsoft Fabric OneLake, Azure Data Lake Storage, Azure Data Explorer lub Lokalny magazyn, musisz określić odwołanie do schematu. Na przykład aby serializować dane do punktu końcowego usługi Microsoft Fabric w formacie delta, należy przekazać schemat do rejestru schematów i odwołać się do niego w konfiguracji docelowego punktu końcowego przepływu danych.

   

2. Wybierz pozycję Kontynuuj , aby skonfigurować miejsce docelowe.

3. Wprowadź wymagane ustawienia dla miejsca docelowego, w tym temat lub tabelę do wysłania danych. Aby uzyskać więcej informacji, zobacz Konfigurowanie miejsca docelowego danych (tematu, kontenera lub tabeli).

Bicep

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

Kubernetes (wersja zapoznawcza)

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

Konfigurowanie miejsca docelowego danych (temat, kontener lub tabela)

Podobnie jak w przypadku źródeł danych, miejsce docelowe danych jest koncepcją używaną do przechowywania punktów końcowych przepływu danych wielokrotnego użytku w wielu przepływach danych. Zasadniczo reprezentuje podkatalog w konfiguracji punktu końcowego przepływu danych. Jeśli na przykład punkt końcowy przepływu danych jest punktem końcowym magazynu, miejsce docelowe danych jest tabelą na koncie magazynu. Jeśli punkt końcowy przepływu danych jest punktem końcowym platformy Kafka, miejsce docelowe danych to temat platformy Kafka.

Typ punktu końcowego	Znaczenie miejsca docelowego danych	opis
MQTT (lub Event Grid)	Temat	Temat MQTT, w którym są wysyłane dane. Obsługiwane są tylko tematy statyczne, bez symboli wieloznacznych.
Kafka (lub Event Hubs)	Temat	Temat platformy Kafka, w którym są wysyłane dane. Obsługiwane są tylko tematy statyczne, bez symboli wieloznacznych. Jeśli punkt końcowy jest przestrzenią nazw usługi Event Hubs, miejsce docelowe danych jest pojedynczym centrum zdarzeń w przestrzeni nazw.
Azure Data Lake Storage	Kontener	Kontener na koncie magazynu. Nie tabela.
Microsoft Fabric OneLake	Tabela lub folder	Odpowiada skonfigurowanemu typowi ścieżki dla punktu końcowego.
Azure Data Explorer	Table	Tabela w bazie danych usługi Azure Data Explorer.
Magazyn lokalny	Folder	Nazwa folderu lub katalogu w instalacji woluminu trwałego magazynu lokalnego. W przypadku korzystania z usługi Azure Container Storage włączonej przez woluminy usługi Azure Arc Cloud Ingest Edge musi być zgodny z parametrem spec.path utworzonego podwolułki.

Aby skonfigurować miejsce docelowe danych:

Portal

W przypadku korzystania ze środowiska operacji pole miejsca docelowego danych jest automatycznie interpretowane na podstawie typu punktu końcowego. Jeśli na przykład punkt końcowy przepływu danych jest punktem końcowym magazynu, na stronie szczegółów miejsca docelowego zostanie wyświetlony monit o wprowadzenie nazwy kontenera. Jeśli punkt końcowy przepływu danych jest punktem końcowym MQTT, strona szczegółów miejsca docelowego wyświetli monit o wprowadzenie tematu itd.

Bicep

Składnia jest taka sama dla wszystkich punktów końcowych przepływu danych:

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

Aby na przykład wysłać dane z powrotem do lokalnego brokera MQTT statycznego tematu MQTT, użyj następującej konfiguracji:

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

Jeśli jednak masz niestandardowy punkt końcowy centrum zdarzeń, konfiguracja będzie wyglądać następująco:

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

Inny przykład użycia punktu końcowego magazynu jako miejsca docelowego:

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

Kubernetes (wersja zapoznawcza)

Składnia jest taka sama dla wszystkich punktów końcowych przepływu danych:

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

Aby na przykład wysłać dane z powrotem do lokalnego brokera MQTT statycznego tematu MQTT, użyj następującej konfiguracji:

destinationSettings:
  endpointRef: default
  dataDestination: example-topic

Jeśli jednak masz niestandardowy punkt końcowy centrum zdarzeń, konfiguracja będzie wyglądać następująco:

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

Inny przykład użycia punktu końcowego magazynu jako miejsca docelowego:

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

Przykład

Poniższy przykład to konfiguracja przepływu danych, która używa punktu końcowego MQTT dla źródła i miejsca docelowego. Źródło filtruje dane z tematu azure-iot-operations/data/thermostatMQTT . Przekształcenie konwertuje temperaturę na Fahrenheit i filtruje dane, w których temperatura pomnożona przez wilgotność jest mniejsza niż 100000. Miejsce docelowe wysyła dane do tematu factoryMQTT .

Portal

Zobacz Karty Bicep lub Kubernetes, aby zapoznać się z przykładem konfiguracji.

Bicep

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

Kubernetes (wersja zapoznawcza)

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

Aby wyświetlić więcej przykładów konfiguracji przepływu danych, zobacz Interfejs API REST platformy Azure — przepływ danych i przewodnik Szybki start Bicep.

Sprawdzanie, czy przepływ danych działa

Wykonaj czynności opisane w samouczku: dwukierunkowy mostek MQTT do usługi Azure Event Grid , aby sprawdzić, czy przepływ danych działa.

Eksportowanie konfiguracji przepływu danych

Aby wyeksportować konfigurację przepływu danych, możesz użyć środowiska operacji lub wyeksportować zasób niestandardowy przepływu danych.

Portal

Wybierz przepływ danych, który chcesz wyeksportować, i wybierz pozycję Eksportuj na pasku narzędzi.

Bicep

Bicep jest infrastrukturą jako kodem i nie jest wymagany żaden eksport. Użyj pliku szablonu Bicep, aby utworzyć przepływ danych, aby szybko skonfigurować i skonfigurować przepływy danych.

Kubernetes (wersja zapoznawcza)

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

Właściwa konfiguracja przepływu danych

Aby upewnić się, że przepływ danych działa zgodnie z oczekiwaniami, sprawdź następujące kwestie:

• Domyślny punkt końcowy przepływu danych MQTT musi być używany jako źródło lub miejsce docelowe.
• Profil przepływu danych istnieje i jest przywołyny w konfiguracji przepływu danych.
• Źródło to punkt końcowy MQTT, punkt końcowy platformy Kafka lub zasób. Punktów końcowych typu magazynu nie można używać jako źródła.
• W przypadku używania usługi Event Grid jako źródła liczba wystąpień profilu przepływu danych jest ustawiona na 1, ponieważ broker MQTT usługi Event Grid nie obsługuje subskrypcji udostępnionych.
• W przypadku używania usługi Event Hubs jako źródła każde centrum zdarzeń w przestrzeni nazw jest osobnym tematem platformy Kafka i musi być określone jako źródło danych.
• Przekształcenie, jeśli jest używane, jest skonfigurowane z właściwą składnią, w tym prawidłowe ucieczki znaków specjalnych.
• W przypadku używania punktów końcowych typu magazynu jako miejsca docelowego zostanie określony schemat.

Następne kroki

• Mapuj dane przy użyciu przepływów danych
• Konwertowanie danych przy użyciu przepływów danych
• Wzbogacanie danych przy użyciu przepływów danych
• Omówienie schematów komunikatów
• Zarządzanie profilami przepływu danych