Azure IoT Operations에서 데이터 흐름 만들기

1. 운영 환경에서 데이터 흐름을 만들려면 데이터 흐름>데이터 흐름 만들기를 선택합니다.

2. 자리 표시자 이름 new-data-flow 를 선택하여 데이터 흐름 속성을 설정합니다. 데이터 흐름의 이름을 입력하고 사용할 데이터 흐름 프로필을 선택합니다. 기본 데이터 흐름 프로필은 기본적으로 선택됩니다. 데이터 흐름 프로필에 대한 자세한 내용은 데이터 흐름 프로필 구성을 참조하세요.

   

   중요합니다

   데이터 흐름을 만들 때만 데이터 흐름 프로필을 선택할 수 있습니다. 데이터 흐름을 만든 후에는 데이터 흐름 프로필을 변경할 수 없습니다. 기존 데이터 흐름의 데이터 흐름 프로필을 변경하려면 원래 데이터 흐름을 삭제하고 새 데이터 흐름 프로필을 사용하여 새 데이터 흐름을 만듭니다.

3. 데이터 흐름 다이어그램에서 항목을 선택하여 데이터 흐름에 대한 원본, 변환 및 대상 엔드포인트를 구성합니다.

   

az iot ops dataflow apply 명령을 사용하여 데이터 흐름을 만들거나 변경합니다.

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

--config-file 매개 변수는 리소스 속성을 포함하는 JSON 구성 파일의 경로 및 파일 이름입니다.

이 예제에서는 사용자의 홈 디렉터리에 저장된 다음 콘텐츠로 명명 data-flow.json 된 구성 파일을 가정합니다.

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

다음은 기본 데이터 흐름 프로필을 사용하여 데이터 흐름을 만들거나 업데이트하는 예제 명령입니다.

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

데이터 흐름 만들기를 시작하려면 Bicep .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
}

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

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

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

Kubernetes 매니페스트 .yaml 파일을 만들어 데이터 흐름 만들기를 시작합니다. 이 예에서는 원본, 변환 및 대상 구성을 포함하는 데이터 흐름의 구조를 보여 줍니다.

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

1. 소스 세부 정보에서 메시지 브로커를 선택합니다.

   

2. 메시지 브로커 원본에 대해 다음 설정을 입력합니다.

   설정	설명
   데이터 흐름 엔드포인트	기본 MQTT 메시지 브로커 엔드포인트를 사용하려면 기본값을 선택합니다.
   항목	들어오는 메시지에 대한 구독 항목 필터입니다. 여러 항목을 추가하려면 항목>행 추가를 사용합니다. 항목에 대한 자세한 내용은 MQTT 또는 Kafka 항목 구성을 참조하세요.
   메시지 스키마	들어오는 메시지를 역직렬화하는 데 사용할 스키마입니다. 데이터 역직렬화를 위한 스키마 지정을 참조하세요.

3. 적용을 선택합니다.

다음은 기본 MQTT broker 엔드포인트에 대한 예제 원본 엔드포인트 구성입니다.

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

메시지 브로커 엔드포인트는 Bicep 파일에 구성됩니다. 예를 들어, 다음 엔드포인트는 데이터 흐름의 원본입니다.

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

예를 들어, 메시지 브로커 엔드포인트와 두 개의 항목 필터를 사용하여 원본을 구성하려면 다음 구성을 사용합니다.

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

자산을 데이터 흐름의 원본으로 사용할 수 있습니다. 자산을 원본으로 사용하는 기능은 운영 환경에서만 사용할 수 있습니다.

1. 소스 세부 정보에서 자산을 선택합니다.

2. 원본 엔드포인트로 사용할 자산을 선택합니다.

3. 계속을 선택합니다.

   선택한 자산에 대한 데이터 포인트 목록이 표시됩니다.

   

4. 자산을 원본 엔드포인트로 사용하려면 적용을 선택합니다.

자산을 원본으로 구성하는 기능은 운영 환경에서만 사용할 수 있습니다.

자산을 원본으로 구성하는 기능은 운영 환경에서만 사용할 수 있습니다.

자산을 원본으로 구성하는 기능은 운영 환경에서만 사용할 수 있습니다.

1. 소스 세부 정보에서 메시지 브로커를 선택합니다.

   

2. 메시지 브로커 원본에 대해 다음 설정을 입력합니다.

   설정	설명
   데이터 흐름 엔드포인트	다시 선택 단추를 사용하여 사용자 지정 MQTT 또는 Kafka 데이터 흐름 엔드포인트를 선택합니다. 자세한 내용은 MQTT 데이터 흐름 엔드포인트 구성 또는 Azure Event Hubs 및 Kafka 데이터 흐름 엔드포인트 구성을 참조하세요.
   항목	들어오는 메시지에 대한 구독 항목 필터입니다. 여러 항목을 추가하려면 항목>행 추가를 사용합니다. 항목에 대한 자세한 내용은 MQTT 또는 Kafka 항목 구성을 참조하세요.
   메시지 스키마	들어오는 메시지를 역직렬화하는 데 사용할 스키마입니다. 데이터 역직렬화를 위한 스키마 지정을 참조하세요.

3. 적용을 선택합니다.

자리 표시자 값을 사용자 지정 엔드포인트 이름과 항목으로 바꿉니다.

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

자리 표시자 값을 사용자 지정 엔드포인트 이름과 항목으로 바꿉니다.

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

자리 표시자 값을 사용자 지정 엔드포인트 이름과 항목으로 바꿉니다.

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

운영 환경 데이터 흐름 소스 세부 정보에서 메시지 브로커를 선택한 다음 항목 필드를 사용하여 들어오는 메시지에 대해 구독할 MQTT 항목 필터를 지정합니다. 행 추가를 선택하고 새 항목을 입력하여 여러 MQTT 항목을 추가할 수 있습니다.

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

와일드카드를 사용한 여러 MQTT 항목 필터의 예:

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

여기에서 와일드카드 +은 thermostats 및 humidifiers 항목 아래의 모든 디바이스를 선택하는 데 사용됩니다. # 와일드카드는 temperature 및 humidity 토픽의 모든 하위 주제에서 모든 센서 메시지를 선택하는 데 사용됩니다.

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

와일드카드를 사용한 여러 MQTT 항목 필터의 예:

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

여기에서 와일드카드 +은 thermostats 및 humidifiers 항목 아래의 모든 디바이스를 선택하는 데 사용됩니다. # 와일드카드는 temperature 및 humidity 토픽의 모든 하위 주제에서 모든 센서 메시지를 선택하는 데 사용됩니다.

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

와일드카드를 사용한 여러 항목 필터의 예:

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

여기에서 와일드카드 +은 thermostats 및 humidifiers 항목 아래의 모든 디바이스를 선택하는 데 사용됩니다. # 와일드카드는 토픽의 temperaturehumidity 모든 하위 항목에서 모든 메시지를 선택하는 데 사용됩니다.

운영 환경 데이터 흐름 소스 세부 정보에서 메시지 브로커를 선택하고 항목 필드를 사용하여 공유 구독 그룹과 항목을 지정합니다.

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

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

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

운영 환경 데이터 흐름 소스 세부 정보에서 메시지 브로커를 선택한 다음 항목 필드를 사용하여 들어오는 메시지에 대해 구독할 Kafka 항목 필터를 지정합니다.

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

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

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

운영 환경 데이터 흐름 소스 세부 정보에서 메시지 브로커를 선택하고 메시지 스키마 필드를 사용하여 스키마를 지정합니다. 먼저 업로드 단추를 사용하여 스키마 파일을 업로드할 수 있습니다. 자세한 내용은 메시지 스키마 이해를 참조하세요.

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

스키마 레지스트리를 사용하여 스키마를 저장한 후 데이터 흐름 구성에서 이를 참조할 수 있습니다.

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

스키마 레지스트리를 사용하여 스키마를 저장한 후 데이터 흐름 구성에서 이를 참조할 수 있습니다.

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

운영 환경에서 데이터 흐름>변환 추가(선택 사항)를 선택합니다.

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

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

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

현재 보강 단계는 운영 환경에서 지원되지 않습니다.

데이터를 보강하려면 데이터 흐름 구성에서 속성을 사용할 builtInTransformationSettings 수 있습니다. 이 datasets 속성은 보강에 사용할 데이터 세트를 지정하는 데 사용됩니다.

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

이 예에서는 원본 데이터의 deviceId 필드를 사용하여 데이터 세트의 asset 필드와 일치하는 방법을 보여 줍니다.

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

예를 들어, 원본 데이터의 deviceId 필드를 사용하여 데이터 세트의 asset 필드와 일치시킬 수 있습니다.

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

1. 변환(선택 사항)에서 필터>추가를 선택합니다.

   

2. 필요한 설정을 입력합니다.

   설정	설명
   필터 조건	원본 데이터의 필드를 기준으로 데이터를 필터링하는 조건.
   설명	필터 조건에 대한 설명을 제공합니다.

   필터 조건 필드에 @을 입력하거나 Ctrl + 스페이스를 선택하여 드롭다운에서 데이터 포인트를 선택합니다.

   @$metadata.user_properties.<property> 또는 @$metadata.topic 형식을 사용하여 MQTT 메타데이터 속성을 입력할 수 있습니다. @$metadata.<header> 형식을 사용하여 $metadata 헤더를 입력할 수도 있습니다. $metadata 구문은 메시지 헤더의 일부인 MQTT 속성에만 필요합니다. 자세한 내용은 필드 참조를 참조하세요.

   조건에서는 원본 데이터의 필드를 사용할 수 있습니다. 예를 들어, @temperature > 20과 같은 필터 조건을 사용하면 온도 필드를 기준으로 20 이하인 데이터를 필터링할 수 있습니다.

3. 적용을 선택합니다.

예를 들어, 원본 데이터의 temperature 필드를 사용하여 데이터를 필터링할 수 있습니다.

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

예를 들어, 원본 데이터의 temperature 필드를 사용하여 데이터를 필터링할 수 있습니다.

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

temperature 필드가 20보다 큰 경우, 데이터는 다음 단계로 전달됩니다. temperature 필드가 20보다 작거나 같으면 데이터가 필터링됩니다.

예를 들어, 원본 데이터의 temperature 필드를 사용하여 데이터를 필터링할 수 있습니다.

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

temperature 필드가 20보다 큰 경우, 데이터는 다음 단계로 전달됩니다. temperature 필드가 20보다 작거나 같으면 데이터가 필터링됩니다.

1. 변환(선택 사항)에서 새 속성>추가를 선택합니다.

   

2. 필요한 설정을 입력합니다.

   설정	설명
   속성 키	새 속성의 키를 입력합니다.
   속성 값	새 속성의 값을 입력합니다.
   설명	새 속성에 대한 설명을 제공합니다.

3. 적용을 선택합니다.

예를 들어, 원본 데이터의 temperature 필드를 사용하여 온도를 섭씨로 변환하고 temperatureCelsius 필드에 저장할 수 있습니다. 또한 컨텍스트화 데이터 세트의 ___location 필드를 사용하여 원본 데이터를 보강할 수도 있습니다.

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

$metadata.user_properties.<property> 또는 $metadata.topic 형식을 사용하여 MQTT 메타데이터 속성에 액세스할 수 있습니다. $metadata.<header> 형식을 사용하여 $metadata 헤더를 입력할 수도 있습니다. 자세한 내용은 필드 참조를 참조하세요.

예를 들어, 원본 데이터의 temperature 필드를 사용하여 온도를 섭씨로 변환하고 temperatureCelsius 필드에 저장할 수 있습니다. 또한 컨텍스트화 데이터 세트의 ___location 필드를 사용하여 원본 데이터를 보강할 수도 있습니다.

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

$metadata.user_properties.<property> 또는 $metadata.topic 형식을 사용하여 MQTT 메타데이터 속성에 액세스할 수 있습니다. $metadata.<header> 형식을 사용하여 $metadata 헤더를 입력할 수도 있습니다. 자세한 내용은 필드 참조를 참조하세요.

예를 들어, 원본 데이터의 temperature 필드를 사용하여 온도를 섭씨로 변환하고 temperatureCelsius 필드에 저장할 수 있습니다. 또한 컨텍스트화 데이터 세트의 ___location 필드를 사용하여 원본 데이터를 보강할 수도 있습니다.

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

1. 변환(선택 사항)에서 제거를 선택합니다.

2. 출력 스키마에서 제거할 데이터 포인트를 선택합니다.

   

3. 적용을 선택합니다.

출력 스키마에서 데이터 포인트를 제거하려면 데이터 흐름 구성에서 속성을 사용할 builtInTransformationSettings 수 있습니다. 이 map 속성은 제거할 데이터 포인트를 지정하는 데 사용됩니다.

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

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

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

운영 환경을 위해 데이터 흐름 엔드포인트 세부 정보에서 스키마와 serialization 서식을 지정합니다. serialization 형식을 지원하는 엔드포인트는 Microsoft Fabric OneLake, Azure Data Lake Storage Gen 2, Azure Data Explorer 및 로컬 스토리지입니다. 예를 들어, 델타 형식으로 데이터를 직렬화하려면 스키마 레지스트리에 스키마를 업로드하고 데이터 흐름 대상 엔드포인트 구성에서 이를 참조해야 합니다.

스키마 레지스트리에 스키마를 업로드하면 데이터 흐름 구성에서 해당 스키마를 참조할 수 있습니다.

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

스키마 레지스트리에 스키마를 업로드하면 데이터 흐름 구성에서 해당 스키마를 참조할 수 있습니다.

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

스키마 레지스트리에 스키마를 업로드하면 데이터 흐름 구성에서 해당 스키마를 참조할 수 있습니다.

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

1. 대상으로 사용할 데이터 흐름 엔드포인트를 선택합니다.

   

   스토리지 엔드포인트에는 serialization을 위한 스키마가 필요합니다. Microsoft Fabric OneLake, Azure Data Lake Storage, Azure Data Explorer 또는 Local Storage 대상 엔드포인트를 선택하는 경우 스키마 참조를 지정해야 합니다. 예를 들어, 델타 형식으로 Microsoft Fabric 엔드포인트에 데이터를 직렬화하려면 스키마 레지스트리에 스키마를 업로드하고 데이터 흐름 대상 엔드포인트 구성에서 이를 참조해야 합니다.

   

2. 계속을 선택하여 대상을 구성합니다.

3. 데이터를 보낼 항목이나 테이블을 포함하여 대상에 필요한 설정을 입력합니다. 자세한 내용은 데이터 대상(항목, 컨테이너 또는 테이블) 구성을 참조하세요.

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

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

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

운영 환경을 사용할 때 데이터 대상 필드는 엔드포인트 형식에 따라 자동으로 해석됩니다. 예를 들어, 데이터 흐름 엔드포인트가 스토리지 엔드포인트인 경우 대상 세부 정보 페이지에서 컨테이너 이름을 입력하라는 메시지가 표시됩니다. 데이터 흐름 엔드포인트가 MQTT 엔드포인트인 경우 대상 세부 정보 페이지에서 항목 등을 입력하라는 메시지가 표시됩니다.

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

예를 들어, 로컬 MQTT 브로커에 정적 MQTT 항목으로 데이터를 다시 보내려면 다음 구성을 사용합니다.

{
  "destinationSettings": {
    "endpointRef": "default",
    "dataDestination": "example-topic"
  }
}

또는 사용자 지정 이벤트 허브 엔드포인트가 있는 경우 구성은 다음과 같습니다.

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

구문은 모든 데이터 흐름 엔드포인트에서 동일합니다.

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

예를 들어, 로컬 MQTT 브로커에 정적 MQTT 항목으로 데이터를 다시 보내려면 다음 구성을 사용합니다.

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

또는 사용자 지정 이벤트 허브 엔드포인트가 있는 경우 구성은 다음과 같습니다.

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

스토리지 엔드포인트를 대상으로 사용하는 또 다른 예:

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

구문은 모든 데이터 흐름 엔드포인트에서 동일합니다.

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

예를 들어, 로컬 MQTT 브로커에 정적 MQTT 항목으로 데이터를 다시 보내려면 다음 구성을 사용합니다.

destinationSettings:
  endpointRef: default
  dataDestination: example-topic

또는 사용자 지정 이벤트 허브 엔드포인트가 있는 경우 구성은 다음과 같습니다.

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

스토리지 엔드포인트를 대상으로 사용하는 또 다른 예:

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

az iot ops dataflow apply 명령을 사용하여 데이터 흐름을 만들거나 변경합니다.

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

--config-file 매개 변수는 리소스 속성을 포함하는 JSON 구성 파일의 경로 및 파일 이름입니다.

이 예제에서는 사용자의 홈 디렉터리에 저장된 다음 콘텐츠로 명명 data-flow.json 된 구성 파일을 가정합니다.

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

다음은 기본 데이터 흐름 프로필을 사용하여 데이터 흐름을 만들거나 업데이트하는 예제 명령입니다.

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

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

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

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

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

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

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

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

내보내려는 데이터 흐름을 선택하고 도구 모음에서 내보내기를 선택합니다.

az iot ops dataflow show 명령을 사용하여 데이터 흐름을 내보냅니다.

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

다음은 이름이 JSON 파일로 명명된 data-flow 데이터 흐름을 내보내는 예제 명령입니다 data-flow.json.

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

Bicep은 코드형 인프라이므로 내보내기가 필요하지 않습니다. Bicep 파일을 사용하여 데이터 흐름을 빠르게 설정하고 구성합니다.

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