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

Ważne

Usługa Azure IoT Operations Preview — włączona przez usługę Azure Arc jest obecnie dostępna w wersji zapoznawczej. Nie należy używać tego oprogramowania w wersji zapoznawczej w środowiskach produkcyjnych.

Po udostępnieniu ogólnie dostępnej wersji należy wdrożyć nową instalację operacji usługi Azure IoT. Nie będzie można uaktualnić instalacji w wersji zapoznawczej.

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ć zaraz po wystąpieniu usługi Azure IoT Operations Preview 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

Profil przepływu danych określa liczbę wystąpień dla przepływów danych, które mają być używane. Jeśli nie potrzebujesz wielu grup przepływów danych z różnymi ustawieniami skalowania, możesz użyć domyślnego profilu przepływu danych. 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-09-15-preview' existing = {
  name: aioInstanceName
}

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

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

// Pointer to the default dataflow profile
resource defaultDataflowProfile 'Microsoft.IoTOperations/instances/dataflowProfiles@2024-09-15-preview' existing = {
  parent: aioInstance
  name: 'default'
}

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

Kubernetes

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

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.

Używanie 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

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.

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

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.

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.

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

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

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

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

Jeśli liczba wystąpień w profilu przepływu danych jest większa niż 1, 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. Jeśli chcesz użyć innego identyfikatora udostępnionej grupy subskrypcji, możesz zastąpić go w temacie, na przykład $shared/mygroup/topic1.

Ważne

Przepływy danych wymagające subskrypcji udostępnionej, gdy liczba wystąpień jest większa niż 1, 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 1 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

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

Określanie schematu do deserializacji danych

Jeśli dane źródłowe mają opcjonalne pola lub pola z różnymi typami, określ schemat deserializacji w celu zapewnienia spójności. Na przykład dane mogą zawierać pola, które nie są obecne we wszystkich komunikatach. Bez schematu transformacja nie może obsłużyć tych pól, ponieważ mają puste wartości. Za pomocą schematu można określić wartości domyślne lub zignorować pola.

Określanie schematu jest istotne tylko w przypadku korzystania ze źródła MQTT lub Kafka. Jeśli źródło jest zasobem, schemat jest automatycznie wnioskowany z definicji zasobu.

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

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

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, zmienianie nazwy lub dodawanie nowej właściwości: dodaj dodatkowe dane do danych źródłowych, biorąc pod uwagę zestaw danych i warunek, aby był zgodny.
2. Filtr: filtruj dane na podstawie warunku.
3. Mapuj lub oblicz: przenieś dane z jednego pola do drugiego z opcjonalną konwersją.

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

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, możesz użyć zestawu danych referencyjnych w rozproszonym magazynie stanów operacji usługi Azure IoT (DSS). 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 usługi DSS przy użyciu przykładowego narzędzia zestawu DSS. Nazwy kluczy w rozproszonym magazynie stanów odpowiadają zestawowi danych w konfiguracji przepływu danych.

Portal

W środowisku operacji etap Wzbogacanie jest obecnie obsługiwany przy użyciu przekształceń Zmień nazwę i Nowa właściwość .

1. W środowisku operacji wybierz przepływ danych, a następnie pozycję Dodaj przekształcenie (opcjonalnie).

2. Wybierz pozycję Zmień nazwę lub Nowe przekształcenia właściwości , a następnie wybierz pozycję Dodaj.

   

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

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. Wybierz punkty danych do uwzględnienia w zestawie danych.

3. Dodaj warunek filtru i opis.

   

4. Wybierz Zastosuj.

Można na przykład użyć warunku filtru, takiego jak temperature > 20 filtrowanie danych mniejszych lub równych 20 na podstawie pola temperatury.

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

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

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

2. Wprowadź wymagane pola i wyrażenia.

   

3. Wybierz Zastosuj.

Bicep

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

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.

Portal

Obecnie określanie schematu danych wyjściowych i serializacji nie jest obsługiwane w środowisku operacji.

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

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ą odwołania do schematu. Jeśli utworzono docelowe punkty końcowe magazynu dla usługi Microsoft Fabric OneLake, ADLS Gen 2, Azure Data Explorer i Local Storage, musisz określić odwołanie do schematu.

Portal

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

   

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

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

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 . Transformacja konwertuje temperaturę na Fahrenheita 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-09-15-preview' existing = {
  name: aioInstanceName
}

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

// Pointer to the default dataflow endpoint
resource defaultDataflowEndpoint 'Microsoft.IoTOperations/instances/dataflowEndpoints@2024-09-15-preview' existing = {
  parent: aioInstance
  name: 'default'
}

// Pointer to the default dataflow profile
resource defaultDataflowProfile 'Microsoft.IoTOperations/instances/dataflowProfiles@2024-09-15-preview' existing = {
  parent: aioInstance
  name: 'default'
}

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

Kubernetes

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

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

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