Azure IoT İşlemleri'nde veri akışı grafı oluşturma

Veri akışı grafiği, kaynaklar ve hedefler arasında hareket ettikçe verileri dönüştüren birleştirilebilir işlem hattıdır. Standart veri akışı sabit zenginleştirme, filtre ve harita dizisini izler. Veri akışı grafiği, dönüşümleri herhangi bir sırada zincirlemenize, paralel yollara dallanmanıza ve zaman içinde verileri toplamanıza olanak tanır.

Bu makalede, adım adım veri akışı grafı oluşturma adımları gösterilir. Veri akışı graflarına ve kullanılabilir dönüşümlere genel bakış için bkz. Veri akışı graflarına genel bakış.

Veri akışı grafikleri şu anda yalnızca MQTT, Kafka ve OpenTelemetry uç noktalarını desteklemektedir. Veri Gölü, Microsoft Fabric OneLake, Azure Veri Gezgini ve Local Storage gibi diğer uç nokta türlerini desteklemiyorlar.

Dönüşümler, değerleri, test koşullarını ve referans alanlarını hesaplamak için bir ifade dili kullanır. İfadeler, girdilere isimle değil, konuma göre atıfta bulunur: listedeki inputs ilk girdi $1, ikinci girdi $2, ve benzeri. cToF gibi yerleşik işlevler, bu değerleri dönüştürür ve işler.

Operatörler, fonksiyonlar, veri tipleri ve metaveri alanlarının tam listesi için İfadeler referansına bakınız.

Önkoşullar

• Kubernetes kümesinde dağıtılan Azure IoT İşlemleri örneği. Daha fazla bilgi için bkz. Deploy Azure IoT İşlemleri.

• Geliştirme makinenizde yüklü Azure CLI. Kullanılabilir Azure CLI uzantıları sayfasında, azure-iot-ops uzantısını kullanmak için gereken en düşük sürüme bakın. Sürümünüzü denetlemek ve az --version gerekirse güncelleştirmek için kullanınaz upgrade. Daha fazla bilgi için bkz. Azure CLI yükleme.

• Azure CLI için Azure IoT İşlemleri uzantısı. Uzantıyı eklemek veya en son sürüme güncelleştirmek için aşağıdaki komutu kullanın:

  az extension add --upgrade --name azure-iot-ops
  

• Azure IoT İşlemleri sürüm 1.2 veya üzeri.

• Veri akışı profili. Varsayılan profili kullanabilirsiniz.

• Kaynağınız ve hedefiniz için bir veri akışı uç noktası . Başlangıç için varsayılan MQTT aracı uç noktası çalışır.

Bu makaledeki Azure CLI örnek, her değeri bir kez ayarlayabilmeniz ve ardından komutları kopyalayıp yapıştırmanız için ortam değişkenlerini kullanır as-is. Eğer hızlı başlangıçta Azure IoT İşlemleri Codespaces ortamını kullanıyorsanız, bu değişkenler zaten sizin için ayarlanmış ve bu adımı atlayabilirsiniz. Aksi takdirde, komutları çalıştırmadan önce shell'inizde aşağıdaki ortam değişkenlerini ayarlayın.

Aşağıdaki betikler en sık kullanılan ortam değişkenlerini ayarlar:

Ortam değişkeni	Açıklama
SUBSCRIPTION_ID	Azure IoT İşlemleri örneğini içeren aboneliğin kimliği.
RESOURCE_GROUP	Azure IoT İşlemleri örneğini içeren kaynak grubunun adı.
AIO_INSTANCE_NAME	Azure IoT İşlemleri örneğinizin adı. Örneklerinizi listelemek için çalıştırın az iot ops list -o table.
CLUSTER_NAME	Azure Arc özellikli Kubernetes kümesinin adı, örneğini barındırıyor.
LOCATION	Yeni kaynaklar için kullanılacak Azure bölgesi, örneğin eastus.

Bash

SUBSCRIPTION_ID=<subscription-id>
RESOURCE_GROUP=<resource-group-name>
AIO_INSTANCE_NAME=<instance-name>
CLUSTER_NAME=<cluster-name>
LOCATION=<region>

PowerShell

$SUBSCRIPTION_ID = "<subscription-id>"
$RESOURCE_GROUP = "<resource-group-name>"
$AIO_INSTANCE_NAME = "<instance-name>"
$CLUSTER_NAME = "<cluster-name>"
$LOCATION = "<region>"

Sadece bu makalenin kullandığı değişkenleri ayarlamanız yeterlidir. Bu makale, seçtiğiniz kaynak adları için ek ortam değişkenleri kullanabilir. Makale, tanıtıldıkları yere nasıl yerleştirileceğini açıklıyor.

Bu makale ayrıca veri akışı grafiği ve profil adları için ve GRAPH_NAME ortam değişkenlerini kullanırPROFILE. İlgili komutları çalıştırmadan önce her birini ayarlayın.

Veri akışı grafı oluşturma

Veri akışı grafı üç tür öğe içerir: verileri getiren kaynaklar , işleyen dönüşümler ve bunları gönderen hedefler . Bunları verilerin akmasını istediğiniz sırayla bağlayın.

İşlem deneyimi

1. Operasyon deneyiminde, Azure IoT İşlemleri örneğinize gidin.

2. Veri akışı grafiği>Veri akışı grafiği oluşturu seçin.

   

3. Veri akışı grafiği için bir ad girin ve bir veri akışı profili seçin. Varsayılan profil varsayılan olarak seçilidir.

   

4. Tuvale öğeler ekleyerek işlem hattınızı oluşturun.

   1. Kaynak ekleme: Kaynak uç noktasını seçin ve gelen iletiler için abone olunacak konuları yapılandırın.

      

   2. Dönüşüm ekleme: Verileri işlemek için bir veya daha fazla dönüşüm seçin. Mevcut dönüşümler arasında map, filter, branch, concatenate, window ve throttle bulunur. Her dönüştürme türüyle ilgili ayrıntılar için bkz. Veri akışı graflarına genel bakış.

      

      

   3. Hedef ekleme: Hedef uç noktayı seçin ve işlenen verileri göndermek için konuyu veya yolu yapılandırın.

      

5. Öğeleri verilerin akmasını istediğiniz sırayla bağlayın.

   

6. Veri akışı grafiğini dağıtmak için Kaydet'i seçin.

Azure CLI

Azure CLI, tüm düğümleri ve bağlantıları içeren tek bir JSON yapılandırma dosyasından bir veri akışı grafiği kullanır. Grafiği oluşturmak veya değiştirmek için kullanın az iot ops dataflowgraph apply . Aşağıdaki örnek sıcaklık verilerini okur, Fahrenheit'e dönüştürür ve hedef konuya gönderir.

Veri akışı grafı özelliklerine sahip bir graph.json dosya oluşturun. Dosyada graph.json , her dönüşümün kurallarını value alanda kaçan bir JSON dizisi olarak depolayın. Her dönüşümün kurallarının okunabilir biçimi için bu dönüştürme türünün nasıl yapılır bölümüne bakın.

{
  "mode": "Enabled",
  "nodes": [
    {
      "nodeType": "Source",
      "name": "sensors",
      "sourceSettings": {
        "endpointRef": "default",
        "dataSources": [
          "telemetry/temperature"
        ]
      }
    },
    {
      "nodeType": "Graph",
      "name": "convert",
      "graphSettings": {
        "registryEndpointRef": "default",
        "artifact": "azureiotoperations/graph-dataflow-map:1.0.0",
        "configuration": [
          {
            "key": "rules",
            "value": "{\"map\":[{\"inputs\":[\"*\"],\"output\":\"*\"},{\"inputs\":[\"temperature\"],\"output\":\"temperature_f\",\"expression\":\"cToF($1)\"}]}"
          }
        ]
      }
    },
    {
      "nodeType": "Destination",
      "name": "output",
      "destinationSettings": {
        "endpointRef": "default",
        "dataDestination": "telemetry/converted"
      }
    }
  ],
  "nodeConnections": [
    {
      "from": {
        "name": "sensors"
      },
      "to": {
        "name": "convert"
      }
    },
    {
      "from": {
        "name": "convert"
      },
      "to": {
        "name": "output"
      }
    }
  ]
}

Yapılandırma dosyasını uygulayın.

az iot ops dataflowgraph apply \
  --name temperature-processing \
  --instance $AIO_INSTANCE_NAME \
  --resource-group $RESOURCE_GROUP \
  --config-file graph.json

Grafik, veri akış profilini default kullanır. Farklı bir profil kullanmak için ekleyin --profile <PROFILE_NAME>.

Biceps

Aşağıdaki yapıya sahip bir Bicep .bicep dosyası oluşturun. Bu örnek, sıcaklık verilerini okuyan, Fahrenheit'e dönüştüren ve bir hedef konuya gönderen bir veri akışı grafiği oluşturur.

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

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

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

resource defaultDataflowProfile 'Microsoft.IoTOperations/instances/dataflowProfiles@2026-03-01' existing = {
  parent: aioInstance
  name: 'default'
}

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2026-03-01' = {
  parent: defaultDataflowProfile
  name: 'temperature-processing'
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    profileRef: 'default'
    mode: 'Enabled'
    nodes: [
      {
        nodeType: 'Source'
        name: 'sensors'
        sourceSettings: {
          endpointRef: 'default'
          dataSources: [
            'telemetry/temperature'
          ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'convert'
        graphSettings: {
          registryEndpointRef: 'default'
          artifact: 'azureiotoperations/graph-dataflow-map:1.0.0'
          configuration: [
            {
              key: 'rules'
              value: '{"map":[{"inputs":["*"],"output":"*"},{"inputs":["temperature"],"output":"temperature_f","expression":"cToF($1)"}]}'
            }
          ]
        }
      }
      {
        nodeType: 'Destination'
        name: 'output'
        destinationSettings: {
          endpointRef: 'default'
          dataDestination: 'telemetry/converted'
        }
      }
    ]
    nodeConnections: [
      {
        from: { name: 'sensors' }
        to: { name: 'convert' }
      }
      {
        from: { name: 'convert' }
        to: { name: 'output' }
      }
    ]
  }
}

Bicep dosyasının dağıtımını gerçekleştirin.

az deployment group create --resource-group $RESOURCE_GROUP --template-file main.bicep

Kubernetes (yalnızca hata ayıklama)

Önemli

Kubernetes dağıtım bildirimlerinin kullanımı üretim ortamlarında desteklenmez ve yalnızca hata ayıklama ve test için kullanılmalıdır.

Aşağıdaki yapıya sahip bir Kubernetes bildirim .yaml dosyası oluşturun. Bu örnek, sıcaklık verilerini okuyan, Fahrenheit'e dönüştüren ve bir hedef konuya gönderen bir veri akışı grafiği oluşturur.

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
  name: temperature-processing
  namespace: azure-iot-operations
spec:
  profileRef: default
  nodes:
    - nodeType: Source
      name: sensors
      sourceSettings:
        endpointRef: default
        dataSources:
          - telemetry/temperature

    - nodeType: Graph
      name: convert
      graphSettings:
        registryEndpointRef: default
        artifact: azureiotoperations/graph-dataflow-map:1.0.0
        configuration:
          - key: rules
            value: |
              {
                "map": [
                  {
                    "inputs": ["*"],
                    "output": "*"
                  },
                  {
                    "inputs": ["temperature"],
                    "output": "temperature_f",
                    "expression": "cToF($1)"
                  }
                ]
              }

    - nodeType: Destination
      name: output
      destinationSettings:
        endpointRef: default
        dataDestination: telemetry/converted

  nodeConnections:
    - from:
        name: sensors
      to:
        name: convert
    - from:
        name: convert
      to:
        name: output

Bildirimi uygulayın:

kubectl apply -f main.yaml

Kaynağı yapılandırma

Kaynak, verilerin işlem hattına nereye girdiğini tanımlar. Bir uç nokta referansını ve bir veya daha fazla konuyu belirtin.

İşlem deneyimi

Veri akışı grafı düzenleyicisinde kaynak öğesini seçin ve yapılandırın:

Setting	Açıklama
Bitiş noktası	Kullanılacak veri akışı uç noktası. Yerel MQTT aracısı için varsayılan'ı seçin.
Konular	Gelen iletiler için abone olunacak bir veya daha fazla konu.

Azure CLI

CLI grafiğin tamamını tek seferde uygular, bu nedenle kaynağı Source yapılandırma dosyanızda bir graph.json düğümü olarak yapılandırın, ardından az iot ops dataflowgraph apply: çalıştırın

{
  "nodeType": "Source",
  "name": "sensors",
  "sourceSettings": {
    "endpointRef": "default",
    "dataSources": [
      "telemetry/temperature",
      "telemetry/humidity"
    ]
  }
}

Biceps

{
  nodeType: 'Source'
  name: 'sensors'
  sourceSettings: {
    endpointRef: 'default'
    dataSources: [
      'telemetry/temperature'
      'telemetry/humidity'
    ]
  }
}

Kubernetes (yalnızca hata ayıklama)

Önemli

Kubernetes dağıtım bildirimlerinin kullanımı üretim ortamlarında desteklenmez ve yalnızca hata ayıklama ve test için kullanılmalıdır.

- nodeType: Source
  name: sensors
  sourceSettings:
    endpointRef: default
    dataSources:
      - telemetry/temperature
      - telemetry/humidity

Dönüşümleri ekle

Kaynak ve hedef arasında işlem verilerini dönüştürür. Her dönüşüm, yerleşik bir artefakta referans verir ve yapılandırma kurallarına sahiptir.

Kullanılabilir yerleşik dönüşümler şunlardır:

Dönüşüm	Yapıt	Açıklama
Harita	azureiotoperations/graph-dataflow-map:1.0.0	Alanları yeniden adlandırın, yeniden yapılandırın, hesaplayıp kopyalayın.
Filtre	azureiotoperations/graph-dataflow-filter:1.0.0	Bir koşulla eşleşen iletileri bırakın.
Dal	azureiotoperations/graph-dataflow-branch:1.0.0	Her iletiyi, bir koşula bağlı olarak true veya false yoluna yönlendirin.
Concatenate	azureiotoperations/graph-dataflow-concatenate:1.0.0	İki veya daha fazla yolu bir arada birleştirin.
Pencere	azureiotoperations/graph-dataflow-window:1.0.0	Belirli bir zaman aralığındaki iletileri toplayın, ardından bunları birleştirin.
Kısıtlama	azureiotoperations/graph-dataflow-throttle:1.0.0	MQTT konu desenine göre mesaj hızını sınırlayın.

İletileri dış verilerle zenginleştirme hakkında daha fazla bilgi için bkz. Dış verilerle zenginleştirme.

İşlem deneyimi

Veri akışı grafı düzenleyicisinde Dönüşüm ekle'yi seçin ve dönüştürme türünü seçin. Görsel düzenleyicisinde kuralları yapılandırın.

Azure CLI

Her dönüşüm, nodeType yapılandırma dosyanızda Graph değeri graph.json olarak ayarlanmış bir düğümdür. Dönüşümün kuralları, sıcaklığı Fahrenheit'e dönüştüren bu harita gibi bir JSON nesnesidir:

{
  "map": [
    {
      "inputs": ["temperature"],
      "output": "temperature_f",
      "expression": "cToF($1)"
    }
  ]
}

Özellik configuration bu kuralları bir dizim olarak alır, bu yüzden JSON kurallarından kaçıp alanı yerleştirir value . Tam grafiği az iot ops dataflowgraph apply ile uygulayın:

{
  "nodeType": "Graph",
  "name": "convert",
  "graphSettings": {
    "registryEndpointRef": "default",
    "artifact": "azureiotoperations/graph-dataflow-map:1.0.0",
    "configuration": [
      {
        "key": "rules",
        "value": "{\"map\":[{\"inputs\":[\"temperature\"],\"output\":\"temperature_f\",\"expression\":\"cToF($1)\"}]}"
      }
    ]
  }
}

İpucu

Kaçışlı dizeyi oluşturmak için, kuralları rules.json gibi bir dosyaya kaydedin, jq -c . rules.json komutunu çalıştırın ve tek satırlık çıktıyı value alanına yapıştırın.

Biceps

Her dönüşüm bir nodeType: 'Graph' düğümdür. configuration özelliği kuralları JSON dizesi olarak geçirir:

{
  nodeType: 'Graph'
  name: 'convert'
  graphSettings: {
    registryEndpointRef: 'default'
    artifact: 'azureiotoperations/graph-dataflow-map:1.0.0'
    configuration: [
      {
        key: 'rules'
        value: '{"map":[{"inputs":["temperature"],"output":"temperature_f","expression":"cToF($1)"}]}'
      }
    ]
  }
}

Kubernetes (yalnızca hata ayıklama)

Önemli

Kubernetes dağıtım bildirimlerinin kullanımı üretim ortamlarında desteklenmez ve yalnızca hata ayıklama ve test için kullanılmalıdır.

Her dönüşüm bir nodeType: Graph düğümdür. configuration özelliği kuralları JSON dizesi olarak geçirir:

- nodeType: Graph
  name: convert
  graphSettings:
    registryEndpointRef: default
    artifact: azureiotoperations/graph-dataflow-map:1.0.0
    configuration:
      - key: rules
        value: |
          {
            "map": [
              {
                "inputs": ["temperature"],
                "output": "temperature_f",
                "expression": "cToF($1)"
              }
            ]
          }

Birden çok dönüşümü zincirleme

İstersiz sayıda dönüşümü zincirleyin. Bunları nodeConnections bölümde verilerin akmasını istediğiniz sırayla bağlayın:

İşlem deneyimi

İşlem sırasını tanımlamak için tuvaldeki dönüşümler arasında bağlantıları sürükleyin.

Azure CLI

nodeConnections yapılandırma dosyanızın graph.json bölümünde işleme sırasını tanımlayın:

"nodeConnections": [
  {
    "from": {
      "name": "sensors"
    },
    "to": {
      "name": "remove-bad-data"
    }
  },
  {
    "from": {
      "name": "remove-bad-data"
    },
    "to": {
      "name": "convert"
    }
  },
  {
    "from": {
      "name": "convert"
    },
    "to": {
      "name": "output"
    }
  }
]

Biceps

nodeConnections: [
  { from: { name: 'sensors' }, to: { name: 'remove-bad-data' } }
  { from: { name: 'remove-bad-data' }, to: { name: 'convert' } }
  { from: { name: 'convert' }, to: { name: 'output' } }
]

Kubernetes (yalnızca hata ayıklama)

Önemli

Kubernetes dağıtım bildirimlerinin kullanımı üretim ortamlarında desteklenmez ve yalnızca hata ayıklama ve test için kullanılmalıdır.

nodeConnections:
  - from: { name: sensors }
    to: { name: remove-bad-data }
  - from: { name: remove-bad-data }
    to: { name: convert }
  - from: { name: convert }
    to: { name: output }

Hedefi yapılandırma

Hedef, işlenen verinin nereye gideceğini tanımlar. Uç nokta referansı ve konu veya yol belirtin.

İşlem deneyimi

Hedef öğeyi seçin ve yapılandırın:

Setting	Açıklama
Bitiş noktası	Veri göndermek için veri akışı uç noktası.
Topic	İşlenen verilerin yayımlandığı konu veya yol.

Azure CLI

Hedefi, Destination yapılandırma dosyanızda graph.json düğümü olarak yapılandırın, ardından grafın tamamını az iot ops dataflowgraph apply ile uygulayın:

{
  "nodeType": "Destination",
  "name": "output",
  "destinationSettings": {
    "endpointRef": "default",
    "dataDestination": "telemetry/processed"
  }
}

Biceps

{
  nodeType: 'Destination'
  name: 'output'
  destinationSettings: {
    endpointRef: 'default'
    dataDestination: 'telemetry/processed'
  }
}

Kubernetes (yalnızca hata ayıklama)

Önemli

Kubernetes dağıtım bildirimlerinin kullanımı üretim ortamlarında desteklenmez ve yalnızca hata ayıklama ve test için kullanılmalıdır.

- nodeType: Destination
  name: output
  destinationSettings:
    endpointRef: default
    dataDestination: telemetry/processed

İleti içeriğine dayalı dinamik konu yönlendirmesi için bkz. İletileri farklı konulara yönlendirme.

Veri akışı grafiğinin çalıştığını doğrulama

Bir veri akış grafiğini dağıttıktan sonra çalıştığını doğrulayın:

İşlem deneyimi

Operasyon deneyiminde, durumunu görmek için veri akış grafiğinizi seçin. İyi durumdaki bir grafik çalışıyor durumunu gösterir.

Azure CLI

Grafiğin ayrıntılarını görüntülemek için kullanın az iot ops dataflowgraph show :

az iot ops dataflowgraph show \
  --name temperature-processing \
  --instance $AIO_INSTANCE_NAME \
  --resource-group $RESOURCE_GROUP

Bir profille ilişkili tüm veri akışı grafiklerini listelemek için kullanın az iot ops dataflowgraph list:

az iot ops dataflowgraph list \
  --instance $AIO_INSTANCE_NAME \
  --resource-group $RESOURCE_GROUP

Biceps

Kaynağın DataflowGraph durumunu denetleyin:

az resource show --resource-group $RESOURCE_GROUP --resource-type Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs --name $GRAPH_NAME --parent instances/$AIO_INSTANCE_NAME/dataflowProfiles/$PROFILE

Kubernetes (yalnızca hata ayıklama)

Önemli

Kubernetes dağıtım bildirimlerinin kullanımı üretim ortamlarında desteklenmez ve yalnızca hata ayıklama ve test için kullanılmalıdır.

kubectl get dataflowgraph temperature-processing -n azure-iot-operations

Pod günlüklerinin hataları olup olmadığını denetleyin:

kubectl logs -l app=dataflow -n azure-iot-operations --tail=50

İlgili içerik

• Veri akışı graflarına genel bakış
• Kaynak yapılandırma
• Hedefi yapılandırma
• Harita ile verileri dönüştürme
• Verileri filtreleme ve yönlendirme
• Zaman içinde verileri toplama
• Dış verilerle zenginleştirme
• Gaz kelebeği verileri
• İfadeler Referansı