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.

W tym artykule pokazano, jak utworzyć przepływ danych przy użyciu przykładu, w tym źródła, przekształcenia i miejsca docelowego.

Wymagania wstępne

• Wystąpienie usługi Azure IoT Operations (wersja zapoznawcza)
• Skonfigurowany profil przepływu danych
• Punkty końcowe przepływu danych. Na przykład utwórz punkt końcowy przepływu danych dla lokalnego brokera MQTT. Tego punktu końcowego można użyć zarówno dla źródła, jak i miejsca docelowego. Możesz też wypróbować inne punkty końcowe, takie 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.

Tworzenie przepływu danych

Po utworzeniu punktów końcowych przepływu danych można ich użyć do utworzenia przepływu danych. Pamiętaj, że przepływ danych składa się z trzech części: źródła, transformacji i miejsca docelowego.

Portal

Aby utworzyć przepływ danych w portalu środowiska operacji, wybierz pozycję Przepływ>danych Utwórz przepływ danych.

Kubernetes

Ogólna struktura konfiguracji przepływu danych jest następująca:

apiVersion: connectivity.iotoperations.azure.com/v1beta1
kind: Dataflow
metadata:
  name: my-dataflow
  namespace: azure-iot-operations
spec:
  profileRef: default
  mode: Enabled
  operations:
    - operationType: Source
      sourceSettings:
        # See source configuration section
    - 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.

Konfigurowanie źródła za pomocą punktu końcowego przepływu danych w celu pobrania danych

Aby skonfigurować źródło przepływu danych, określ odwołanie do punktu końcowego i źródło danych. Możesz określić listę źródeł danych dla punktu końcowego.

Portal

Używanie elementu zawartości jako źródła

Zasób można użyć jako źródła przepływu danych. Jest to dostępne tylko w portalu środowiska 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.

Kubernetes

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

Używanie MQTT jako źródła

Portal

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

2. Wprowadź temat MQTT, który chcesz nasłuchiwać dla komunikatów przychodzących.

3. Wybierz schemat komunikatu z listy rozwijanej lub przekaż nowy schemat. 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.

   

4. Wybierz Zastosuj.

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: mq
  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 Ponowne używanie punktów końcowych przepływu danych.

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.

spec:
  operations:
  - operationType: Source
    sourceSettings:
      serializationFormat: Json
      schemaRef: aio-sr://exampleNamespace/exampleAvroSchema:1.0.0

Aby określić schemat, utwórz plik i zapisz go w rejestrze schematów.

{
  "type": "record",
  "name": "Temperature",
  "fields": [
    {"name": "deviceId", "type": "string"},
    {"name": "temperature", "type": "float"}
  ]
}

Uwaga

Jedynym obsługiwanym formatem serializacji jest JSON. Schemat jest opcjonalny.

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

Subskrypcje udostępnione

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

sourceSettings:
  dataSources:
    - $shared/myGroup/thermostats/+/telemetry/temperature/#

Uwaga

Jeśli liczba wystąpień w profilu przepływu danych jest większa niż 1, należy użyć tematu subskrypcji udostępnionej.

Konfigurowanie przekształcania w celu przetwarzania danych

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. Mapa: Przenieś dane z jednego pola do drugiego z opcjonalną konwersją.

Portal

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

Kubernetes

builtInTransformationSettings:
  datasets:
    # ...
  filter:
    # ...
  map:
    # ...

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.

Nazwy kluczy w rozproszonym magazynie stanów odpowiadają zestawowi danych w konfiguracji przepływu danych.

Portal

Obecnie operacja wzbogacania nie jest dostępna w portalu środowiska operacji.

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 thermostat1 polem mają location pola i manufacturer dostępne w filter etapach i map .deviceId

Przykładowe dane można załadować do usługi DSS przy użyciu przykładowego narzędzia zestawu DSS.

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.

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 portalu środowiska 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.

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
      output: temperatureCelsius
      expression: "($1 - 32) * 5/9"
    - 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. Pamiętaj, że punkty końcowe magazynu, takie jak Microsoft Fabric lub Azure Data Lake, wymagają schematu w celu zapewnienia spójności danych.

Portal

Określ schemat danych wyjściowych podczas dodawania docelowego punktu końcowego przepływu danych.

Kubernetes

builtInTransformationSettings:
  serializationFormat: Parquet
  schemaRef: aio-sr://<NAMESPACE>/<SCHEMA>:<VERSION>

Aby określić schemat, możesz utworzyć zasób niestandardowy schematu z definicją schematu.

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

{
  "type": "record",
  "name": "Temperature",
  "fields": [
    {"name": "deviceId", "type": "string"},
    {"name": "temperatureCelsius", "type": "float"}
    {"name": "location", "type": "string"}
  ]
}

Obsługiwane formaty serializacji to JSON, Parquet i Delta.

Konfigurowanie miejsca docelowego przy użyciu punktu końcowego przepływu danych w celu wysyłania danych

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, które są tematami MQTT lub Kafka.

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. Dodaj szczegóły mapowania na podstawie typu miejsca docelowego.

Kubernetes

Aby na przykład skonfigurować miejsce docelowe przy użyciu punktu końcowego MQTT utworzonego wcześniej i statycznego tematu MQTT, użyj następującej konfiguracji:

destinationSettings:
  endpointRef: mq
  dataDestination: factory

Jeśli utworzono punkty końcowe magazynu, takie jak Microsoft Fabric, użyj pola docelowego danych, aby określić nazwę tabeli lub kontenera:

destinationSettings:
  endpointRef: adls
  dataDestination: telemetryTable

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 tematów thermostats/+/telemetry/temperature/# MQTT i humidifiers/+/telemetry/humidity/#. Przekształcenie konwertuje temperaturę na Fahrenheita i filtruje dane, w których temperatura jest mniejsza niż 100000. Miejsce docelowe wysyła dane do tematu factoryMQTT .

apiVersion: connectivity.iotoperations.azure.com/v1beta1
kind: Dataflow
metadata:
  name: my-dataflow
  namespace: azure-iot-operations
spec:
  profileRef: default
  mode: Enabled
  operations:
    - operationType: Source
      sourceSettings:
        endpointRef: mq
        dataSources:
          - thermostats/+/telemetry/temperature/#
          - humidifiers/+/telemetry/humidity/#
    - operationType: builtInTransformation
      builtInTransformationSettings:
        filter:
          - inputs:
              - 'temperature.Value'
              - '"Tag 10".Value'
            expression: "$1*$2<100000"
        map:
          - inputs:
              - '*'
            output: '*'
          - inputs:
              - temperature.Value
            output: TemperatureF
            expression: cToF($1)
          - inputs:
              - '"Tag 10".Value'
            output: 'Tag 10'
    - operationType: Destination
      destinationSettings:
        endpointRef: mq
        dataDestination: factory

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ć portalu ś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.

Kubernetes

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