Azure IoT İşlemlerinde veri akışları oluşturma

Önemli

Bu sayfa, ÖNIZLEME aşamasında olan Kubernetes dağıtım bildirimlerini kullanarak Azure IoT İşlemleri bileşenlerini yönetme yönergelerini içerir. Bu özellik çeşitli sınırlamalarla sağlanır ve üretim iş yükleri için kullanılmamalıdır.

Beta veya önizleme aşamasında olan ya da başka bir şekilde henüz genel kullanıma sunulmamış olan Azure özelliklerinde geçerli olan yasal koşullar için bkz. Microsoft Azure Önizlemeleri için Ek Kullanım Koşulları.

Veri akışı, isteğe bağlı dönüştürmelerle verilerin kaynaktan hedefe aldığı yoldur. Veri akışını özelleştirilmiş Veri Akışı kaynağı oluşturarak veya işlemler deneyimi web arayüzünü kullanarak yapılandırabilirsiniz. Veri akışı üç bölümden oluşur: kaynak, dönüştürme ve hedef.

Kaynak ve hedefi tanımlamak için veri akışı uç noktalarını yapılandırmanız gerekir. Dönüştürme isteğe bağlıdır ve verileri zenginleştirme, verileri filtreleme ve verileri başka bir alana eşleme gibi işlemleri içerebilir.

Önemli

Her veri akışının kaynak veya hedef olarak Azure IoT İşlemleri yerel MQTT aracısı varsayılan uç noktasına sahip olması gerekir.

Veri akışı oluşturmak için Azure IoT İşlemleri'ndeki işlem deneyimini kullanabilirsiniz. İşlem deneyimi, veri akışını yapılandırmak için görsel bir arabirim sağlar. Bicep dosyasını kullanarak veri akışı oluşturmak için Bicep'i veya YAML dosyası kullanarak veri akışı oluşturmak için Kubernetes'i de kullanabilirsiniz.

Kaynağı, dönüşümü ve hedefi yapılandırmayı öğrenmek için okumaya devam edin.

Önkoşullar

Varsayılan veri akışı profilini ve uç noktasını kullanarak Azure IoT İşlemlerinin bir örneğine sahip olduğunuz anda veri akışlarını dağıtabilirsiniz. Ancak, veri akışını özelleştirmek için veri akışı profillerini ve uç noktalarını yapılandırmak isteyebilirsiniz.

Veri akışı profili

Veri akışlarınız için farklı ölçeklendirme ayarlarına ihtiyacınız yoksa Azure IoT İşlemleri tarafından sağlanan varsayılan veri akışı profilini kullanın. Çok fazla veri akışını tek bir veri akışı profiliyle ilişkilendirmekten kaçınmanız gerekir. Çok sayıda veri akışınız varsa, veri akışı profili yapılandırma boyutu sınırı olan 70'i aşma riskini azaltmak için bunları birden çok veri akışı profiline dağıtın.

Yeni bir veri akışı profilini yapılandırmayı öğrenmek için bkz. Veri akışı profillerini yapılandırma.

Veri akışı uç noktaları

Veri akışı için kaynak ve hedefi yapılandırmak için veri akışı uç noktaları gereklidir. Hızlı bir şekilde başlamak için yerel MQTT aracısı için varsayılan veri akışı uç noktasını kullanabilirsiniz. Kafka, Event Hubs, OpenTelemetry veya Azure Data Lake Storage gibi başka veri akışı uç noktaları da oluşturabilirsiniz. Her veri akışı uç noktası türünü yapılandırmayı öğrenmek için bkz . Veri akışı uç noktalarını yapılandırma.

Kullanmaya başlayın

Önkoşulları aldıktan sonra bir veri akışı oluşturmaya başlayabilirsiniz.

İşlem deneyimi

1. İşlemler deneyiminde veri akışı oluşturmak için Veri akışı>Veri akışı oluştur'u seçin.

2. Veri akışı özelliklerini ayarlamak için new-data-flow yer tutucu adını seçin. Veri akışının adını girin ve kullanılacak veri akışı profilini seçin. Varsayılan veri akışı profili varsayılan olarak seçilidir. Veri akışı profilleri hakkında daha fazla bilgi için bkz. Veri akışı profilini yapılandırma.

   

   Önemli

   Veri akışı profilini yalnızca veri akışı oluştururken seçebilirsiniz. Veri akışı oluşturulduktan sonra veri akışı profilini değiştiremezsiniz. Mevcut bir veri akışının veri akışı profilini değiştirmek istiyorsanız, özgün veri akışını silin ve yeni veri akışı profiliyle yeni bir tane oluşturun.

3. Veri akışı diyagramındaki öğeleri seçerek veri akışı için kaynak, dönüştürme ve hedef uç noktayı yapılandırın.

   

Azure CLI

Veri akışı oluşturmak veya değiştirmek için az iot ops dataflow apply komutunu kullanın.

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

--config-file parametresi, kaynak özelliklerini içeren bir JSON yapılandırma dosyasının yolu ve dosya adıdır.

Bu örnekte, kullanıcının giriş dizininde depolanan aşağıdaki içeriğe sahip adlı data-flow.json bir yapılandırma dosyası varsayın:

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

Varsayılan veri akışı profilini kullanarak veri akışı oluşturmak veya güncelleştirmek için örnek bir komut aşağıda verilmişti:

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

Biceps

Veri akışı oluşturmaya başlamak için bir Bicep .bicep dosyası oluşturun. Bu örnekte kaynak, dönüştürme ve hedef yapılandırmalarını içeren veri akışının yapısı gösterilmektedir.

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 (önizleme)

Veri akışı oluşturmaya başlamak için bir Kubernetes bildirim .yaml dosyası oluşturun. Bu örnekte kaynak, dönüştürme ve hedef yapılandırmalarını içeren veri akışının yapısı gösterilmektedir.

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

Veri akışının işlem türlerini yapılandırmayı öğrenmek için aşağıdaki bölümleri gözden geçirin.

Kaynak

Veri akışı için bir kaynak yapılandırmak için uç nokta başvuruyu ve uç noktanın veri kaynaklarının listesini belirtin. Veri akışının kaynağı olarak aşağıdaki seçeneklerden birini belirleyin.

Varsayılan uç nokta kaynak olarak kullanılmıyorsa hedef olarak kullanılmalıdır. Yerel MQTT aracı uç noktasını kullanma hakkında daha fazla bilgi edinmek için bkz. Veri akışları yerel MQTT aracı uç noktasını kullanmalıdır.

1. Seçenek: Varsayılan ileti aracısı uç noktasını kaynak olarak kullanma

İşlem deneyimi

1. Kaynak ayrıntıları'nın altında İleti aracısı'yı seçin.

   

2. İleti aracısı kaynağı için aşağıdaki ayarları girin:

   Ayarlar	Açıklama
   Veri akışı uç noktası	Varsayılan MQTT ileti aracısı uç noktasını kullanmak için varsayılan'ı seçin.
   Konu	Gelen iletiler için abone olunacak konu filtreleme mekanizması. Birden çok konu başlığı eklemek için Konu Başlıkları Konular hakkında daha fazla bilgi için bkz . MQTT veya Kafka konularını yapılandırma.
   İleti şeması	Gelen iletileri seri durumdan çıkarmak için kullanılacak şema. Verileri seri durumdan çıkarmak için şema belirtme konusuna bakın.

3. Uygula’yı seçin.

Azure CLI

Aşağıda, varsayılan MQTT aracı uç noktası için örnek bir kaynak uç nokta yapılandırması verilmişti:

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

Biceps

İleti aracısı uç noktası Bicep dosyasında yapılandırılır. Örneğin, aşağıdaki uç nokta veri akışı için bir kaynaktır.

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

Kubernetes (önizleme)

Örneğin, bir ileti aracısı uç noktası ve iki konu filtresi kullanarak bir kaynak yapılandırmak için aşağıdaki yapılandırmayı kullanın:

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

dataSources Uç nokta yapılandırmasını değiştirmeden MQTT veya Kafka konularını belirtmenize izin verdiğinden, konular farklı olsa bile birden çok veri akışı için uç noktayı yeniden kullanabilirsiniz. Daha fazla bilgi için bkz . Veri kaynaklarını yapılandırma.

2. Seçenek: Varlığı kaynak olarak kullanma

İşlem deneyimi

Veri akışının kaynağı olarak bir varlığı kullanabilirsiniz. Bir varlığı kaynak olarak kullanmak yalnızca işlem deneyiminde kullanılabilir.

1. Kaynak ayrıntıları'nın altında Varlık'ı seçin.

2. Kaynak uç nokta olarak kullanmak istediğiniz varlığı seçin.

3. Devam Et'i seçin.

   Seçili varlık için veri noktalarının listesi görüntülenir.

   

4. Varlığı kaynak uç nokta olarak kullanmak için Uygula'yı seçin.

Azure CLI

Bir varlığı kaynak olarak yapılandırmak yalnızca işlem deneyiminde kullanılabilir.

Biceps

Bir varlığı kaynak olarak yapılandırmak yalnızca işlem deneyiminde kullanılabilir.

Kubernetes (önizleme)

Bir varlığı kaynak olarak yapılandırmak yalnızca işlem deneyiminde kullanılabilir.

Kaynak olarak bir varlık kullanılırken, varlık tanımı veri akışının şemasını çıkarsamak için kullanılır. Varlık tanımı, varlığın veri noktalarının şemasını içerir. Daha fazla bilgi edinmek için bkz . Varlık yapılandırmalarını uzaktan yönetme.

Yapılandırıldıktan sonra, varlıktan alınan veriler yerel MQTT aracısı aracılığıyla veri akışına iletilir. Bu nedenle, kaynak olarak bir varlık kullanıldığında, veri akışı gerçek olarak kaynak olarak yerel MQTT aracısı varsayılan uç noktasını kullanır.

3. Seçenek: Kaynak olarak özel MQTT veya Kafka veri akışı uç noktasını kullanma

Özel bir MQTT veya Kafka veri akışı uç noktası oluşturduysanız (örneğin, Event Grid veya Event Hubs ile kullanmak için), bunu veri akışının kaynağı olarak kullanabilirsiniz. Data Lake veya Fabric OneLake gibi depolama türü uç noktalarının kaynak olarak kullanılamayacağını unutmayın.

İşlem deneyimi

1. Kaynak ayrıntıları'nın altında İleti aracısı'yı seçin.

   

2. İleti aracısı kaynağı için aşağıdaki ayarları girin:

   Ayarlar	Açıklama
   Veri akışı uç noktası	Özel bir MQTT veya Kafka veri akışı uç noktası seçmek için Yeniden Seç düğmesini kullanın. Daha fazla bilgi için bkz . MQTT veri akışı uç noktalarını yapılandırma veya Azure Event Hubs ve Kafka veri akışı uç noktalarını yapılandırma.
   Konu	Gelen iletiler için abone olunacak konu filtreleme mekanizması. Birden çok konu başlığı eklemek için Konu Başlıkları Konular hakkında daha fazla bilgi için bkz . MQTT veya Kafka konularını yapılandırma.
   İleti şeması	Gelen iletileri seri durumdan çıkarmak için kullanılacak şema. Verileri seri durumdan çıkarmak için şema belirtme konusuna bakın.

3. Uygula’yı seçin.

Azure CLI

Yer tutucu değerlerini özel uç nokta adınız ve konularınızla değiştirin.

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

Biceps

Yer tutucu değerlerini özel uç nokta adınız ve konularınızla değiştirin.

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

Kubernetes (önizleme)

Yer tutucu değerlerini özel uç nokta adınız ve konularınızla değiştirin.

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

Veri kaynaklarını yapılandırma (MQTT veya Kafka konuları)

Veri akışı uç noktası yapılandırmasını değiştirmenize gerek kalmadan bir kaynakta birden çok MQTT veya Kafka konusu belirtebilirsiniz. Bu esneklik, konular farklılık gösterse bile aynı uç noktanın birden çok veri akışında yeniden kullanılabilmesi anlamına gelir. Daha fazla bilgi için bkz . Veri akışı uç noktalarını yeniden kullanma.

MQTT konuları

Kaynak bir MQTT (Event Grid dahil) uç noktası olduğunda, gelen iletilere abone olmak için MQTT konu filtresi kullanabilirsiniz. Konu filtresi, birden çok konuya abone olmak için joker karakterler içerebilir. Örneğin, thermostats/+/sensor/temperature/# termostatlardan gelen tüm sıcaklık sensörü iletilerine abone olur. MQTT konu filtrelerini yapılandırmak için:

İşlem deneyimi

İşlem deneyimi veri akışı Kaynak ayrıntıları bölümünde İleti aracısı'nı seçin, ardından Konu(lar) alanını kullanarak gelen iletilere abone olunacak MQTT konu filtrelerini belirtin. Satır ekle'yi seçip yeni bir konu girerek birden çok MQTT konusu ekleyebilirsiniz.

Azure CLI

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

Joker karakterler içeren birden çok MQTT konu filtresi içeren örnek:

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

Burada + joker karakteri, thermostats ve humidifiers konuları altındaki tüm cihazları seçmek için kullanılır. # ve temperature konularının tüm alt konularındaki tüm algılayıcı iletilerini seçmek için joker humidity karakteri kullanılır.

Biceps

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

Joker karakterler içeren birden çok MQTT konu filtresi içeren örnek:

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

Burada + joker karakteri, thermostats ve humidifiers konuları altındaki tüm cihazları seçmek için kullanılır. # ve temperature konularının tüm alt konularındaki tüm algılayıcı iletilerini seçmek için joker humidity karakteri kullanılır.

Kubernetes (önizleme)

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

Joker karakter içeren birden çok konu filtresi içeren örnek:

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

Burada + joker karakteri, thermostats ve humidifiers konuları altındaki tüm cihazları seçmek için kullanılır. Joker # karakteri, temperature ve humidity konularının tüm alt konularındaki tüm iletileri seçmek için kullanılır.

Paylaşılan abonelikler

Paylaşılan abonelikleri ileti aracısı kaynaklarıyla kullanmak için, paylaşılan abonelik konusunu biçiminde $shared/<GROUP_NAME>/<TOPIC_FILTER>belirtebilirsiniz.

İşlem deneyimi

İşlem deneyimi veri akışı Kaynak ayrıntıları bölümünde İleti aracısı'nı seçin ve Konu alanını kullanarakpaylaşılan abonelik grubunu ve konuyu belirtin.

Azure CLI

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

Biceps

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

Kubernetes (önizleme)

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

Veri akışı profilindeki örnek sayısı birden büyükse, paylaşılan abonelik ileti aracısı kaynağı kullanan tüm veri akışları için otomatik olarak etkinleştirilir. Bu durumda ön $shared ek eklenir ve paylaşılan abonelik grubu adı otomatik olarak oluşturulur. Örneğin, örnek sayısı 3 olan bir veri akışı profiliniz varsa ve veri akışınız konu başlıkları topic1 ve topic2 ile yapılandırılmışsa, kaynak olarak bir ileti aracısı uç noktası kullanıyorsa, bunlar otomatik olarak $shared/<GENERATED_GROUP_NAME>/topic1 ve $shared/<GENERATED_GROUP_NAME>/topic2 şeklinde paylaşılan aboneliklere dönüştürülür.

Yapılandırmanızda adlı $shared/mygroup/topic bir konuyu açıkça oluşturabilirsiniz. Ancak, $shared ön eki gerektiğinde otomatik olarak eklendiğinden $shared konuyu açıkça eklemek önerilmez. Veri akışları, ayarlanmadıysa grup adıyla optimizasyon yapabilir. Örneğin, $share ayarlanmadı ve veri akışlarının yalnızca konu adı üzerinde çalışması gerekir.

Önemli

Örnek sayısı birden fazla olduğunda paylaşılan abonelik gerektiren veri akışları, paylaşılan abonelikleri desteklemediğinden kaynak olarak Event Grid MQTT aracısını kullanırken önemlidir. Eksik iletileri önlemek için kaynak olarak Event Grid MQTT aracısını kullanırken veri akışı profil örneği sayısını bir olarak ayarlayın. Veri akışının abone olduğu ve buluttan mesaj aldığı an budur.

Kafka konuları

Kaynak bir Kafka (Event Hubs dahil) uç noktası olduğunda, gelen iletiler için abone olunacak tek tek Kafka konularını belirtin. Joker karakterler desteklenmez, bu yüzden her konuyu tek tek ve statik olarak belirtmeniz gerekir.

Not

Event Hubs'ı Kafka uç noktası üzerinden kullanırken, ad alanındaki her bir olay hub'ı Kafka konusu olarak işlev görür. Örneğin, iki olay hub'ı thermostats ve humidifiers içeren bir Event Hubs ad alanınız varsa, her bir olay hub'unu bir Kafka konusu olarak belirtebilirsiniz.

Kafka konularını yapılandırmak için:

İşlem deneyimi

İşlem deneyimi veri akışı Kaynak ayrıntıları bölümünde İleti aracısı'nı seçin, ardından Konu alanını kullanarak gelen iletilere abone olunacak Kafka konu filtresi belirtin.

Not

İşlem deneyiminde yalnızca bir konu filtresi belirtilebilir. Birden çok konu filtresi kullanmak için Bicep veya Kubernetes kullanın.

Azure CLI

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

Biceps

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

Kubernetes (önizleme)

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

Kaynak şemayı belirtme

Kaynak olarak MQTT veya Kafka kullanırken, işlem deneyimi web kullanıcı arabirimindeki veri noktalarının listesini görüntülemek için bir şeması belirtebilirsiniz. Gelen iletileri seri durumdan çıkarmak ve doğrulamak için şema kullanmak şu anda desteklenmemektedir.

Kaynak bir varlıksa, şema varlık tanımından otomatik olarak çıkarılır.

İpucu

Şemayı örnek bir veri dosyasından oluşturmak için Şema Nesil Yardımcısı'nı kullanın.

Bir kaynaktan gelen iletileri seri durumdan çıkarmak için kullanılan şemayı yapılandırmak için:

İşlem deneyimi

İşlem deneyimi veri akışı Kaynak ayrıntıları bölümünde İleti aracısı'nı seçin ve şemayı belirtmek için İleti şeması alanını kullanın. Önce bir şema dosyasını yüklemek için Karşıya Yükle düğmesini kullanabilirsiniz. Daha fazla bilgi edinmek için bkz . İleti şemalarını anlama.

Azure CLI

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

Biceps

Şemayı depolamak için şema kayıt defterini kullandıktan sonra, buna veri akışı yapılandırmasında başvurabilirsiniz.

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

Kubernetes (önizleme)

Şemayı depolamak için şema kayıt defterini kullandıktan sonra, buna veri akışı yapılandırmasında başvurabilirsiniz.

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

Daha fazla bilgi edinmek için bkz . İleti şemalarını anlama.

Disk kalıcılığı isteme

İstek diski kalıcılığı, veri akışlarının yeniden başlatmalar arasında durumu korumasına olanak tanır. Bu özelliği etkinleştirdiğinizde, bağlı komisyoncu yeniden başlatılırsa grafiğin işlem durumunu geri yükler. Bu özellik, ara verileri kaybetmenin sorun olduğu durum bilgisi tutan işleme senaryoları için kullanışlıdır. İstek diski kalıcılığını etkinleştirdiğinizde, aracı abone kuyruğundaki iletiler gibi MQTT verilerini diskte kalıcı hale gelir. Bu yaklaşım, güç kesintileri veya aracı yeniden başlatmaları sırasında veri akışınızın veri kaynağının veri kaybetmemesini sağlar. Veri akışı başına kalıcılık yapılandırıldığından aracı en iyi performansı korur, bu nedenle yalnızca kalıcılık gerektiren veri akışları bu özelliği kullanır.

Veri akışı grafiği, bir MQTTv5 kullanıcı özelliği kullanarak abonelik sırasında bu kalıcılığı istemektedir. Bu özellik yalnızca aşağıdaki durumlarda çalışır:

• Veri akışı, kaynak olarak MQTT aracısını veya varlığını kullanır
• MQTT aracısı, abone kuyrukları gibi veri türleri için Enabled olarak ayarlanmış dinamik kalıcılık moduyla kalıcılık özelliğini etkinleştirmiştir.

Bu yapılandırma, veri akışları gibi MQTT istemcilerinin MQTTv5 kullanıcı özelliklerini kullanarak abonelikleri için disk kalıcılığı istemesine olanak tanır. MQTT aracısı kalıcılık yapılandırması hakkında ayrıntılı bilgi için bkz. MQTT aracısı kalıcılığını yapılandırma.

Ayar Enabled veya Disabled kabul eder. Disabled varsayılan değerdir.

İşlem deneyimi

Bir veri akışı oluşturduğunuzda veya düzenlerken Düzenle'yi ve ardından Veri kalıcılığı iste'nin yanındaki Evet'i seçin.

Azure CLI

requestDiskPersistence Veri akışı yapılandırma dosyanıza özelliğini ekleyin:

{
    "mode": "Enabled",
    "requestDiskPersistence": "Enabled",
    "operations": [
        // ... your data flow operations
    ]
}

Biceps

requestDiskPersistence veri akışı kaynağınıza özelliğini ekleyin. API sürümü 2025-10-01 veya daha yeni:

resource dataflow 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflows@2025-10-01' = {
  parent: defaultDataflowProfile
  name: dataflowName
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    mode: 'Enabled'
    requestDiskPersistence: 'Enabled'
    operations: [
      // ... your data flow operations
    ]
  }
}

Kubernetes (önizleme)

requestDiskPersistence Özelliğini veri akışı belirtiminize ekleyin. API sürümü veya üstü:connectivity.iotoperations.azure.com/v1beta1

apiVersion: connectivity.iotoperations.azure.com/v1beta1
kind: Dataflow
metadata:
  name: <DATAFLOW_NAME>
  namespace: azure-iot-operations
spec:
  profileRef: default 
  mode: Enabled
  requestDiskPersistence: Enabled
  operations:
    # ... your data flow operations

Dönüşüm

Dönüştürme işlemi, verileri hedefe göndermeden önce kaynaktan dönüştürebileceğiniz yerdir. Dönüştürmeler isteğe bağlıdır. Verilerde değişiklik yapmanız gerekmiyorsa, dönüştürme işlemini veri akışı yapılandırmasına eklemeyin. Birden çok dönüşüm, yapılandırmada belirtildiği sıraylan bağımsız olarak aşamalar halinde zincirlenir. Aşamaların sırası her zaman:

1. Zenginleştirme: Eşleştirecek bir veri kümesi ve koşul verilip kaynak verilere ek veriler ekleyin.
2. Filtre: Verileri bir koşula göre filtreleyin.
3. Eşleme, İşlem, Yeniden Adlandırma veya Yeni özellik ekleme: İsteğe bağlı dönüştürmeyle verileri bir alandan diğerine taşıyın.

Bu bölüm, veri akışı dönüşümlerine giriş niteliğindedir. Daha ayrıntılı bilgi için bkz. Veri akışlarını kullanarak verileri eşleme, Veri akışı dönüştürmelerini kullanarak verileri dönüştürme ve Veri akışlarını kullanarak verileri zenginleştirme.

İşlem deneyimi

İşlem deneyiminde Veri akışı

Azure CLI

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

Biceps

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

Kubernetes (önizleme)

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

Zenginleştir: Başvuru verilerini ekle

Verileri zenginleştirmek için öncelikle Azure IoT İşlemleri durum deposuna başvuru veri kümesi ekleyin. Veri kümesi, bir koşula göre kaynak verilere ek veri eklemek için kullanılır. Koşul, kaynak verilerde veri kümesindeki bir alanla eşleşen bir alan olarak belirtilir.

Durum deposu CLI'sini kullanarak örnek verileri durum deposuna yükleyebilirsiniz. Durum deposundaki anahtar adları, veri akışı yapılandırmasındaki bir veri kümesine karşılık gelir.

İşlem deneyimi

Şu anda Zenginleştirme aşaması operasyon deneyiminde desteklenmemekte.

Azure CLI

Verileri zenginleştirmek için veri akışı yapılandırmasında özelliğini kullanabilirsiniz builtInTransformationSettings . datasets özelliği, zenginleştirme için kullanılacak veri kümelerini belirtmek için kullanılır.

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

Biceps

Bu örnek, kaynak verilerdeki deviceId alanını, veri kümesindeki asset alanıyla nasıl eşleştirebileceğinizi göstermektedir.

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

Kubernetes (önizleme)

Örneğin, kaynak verilerdeki deviceId alanını, veri kümesindeki asset alanı ile eşleştirmek için kullanabilirsiniz.

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

Veri kümesinin asset alanıyla aşağıdakine benzer bir kaydı varsa:

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

deviceId alanı thermostat1 ile eşleşen kaynaktaki veriler, location ve manufacturer alanlarına sahiptir ve bu alanlar filtreleme ve haritalama aşamalarında kullanılabilir.

Koşul söz dizimi hakkında daha fazla bilgi için bkz. Veri akışlarını kullanarak verileri zenginleştirme ve Veri akışlarını kullanarak verileri dönüştürme.

Filtre: Bir koşula göre verileri filtreleme

Bir koşuldaki verileri filtrelemek için aşamayı filter kullanabilirsiniz. Koşul, kaynak verilerde bir değerle eşleşen bir alan olarak belirtilir.

İşlem deneyimi

1. Dönüştür (isteğe bağlı) altında Filtre

   

2. Gerekli ayarları girin.

   Ayarlar	Açıklama
   Filtre koşulu	Kaynak verilerdeki bir alana göre verileri filtreleme koşulu.
   Açıklama	Filtre koşulu için bir açıklama sağlayın.

   Filtre koşulu alanına @ girin veya açılır listeden veri noktaları seçmek için Ctrl + Space tuşlarına basın.

   veya @$metadata.user_properties.<property>biçimini @$metadata.topic kullanarak MQTT meta veri özelliklerini girebilirsiniz. biçimini @$metadata.<header>kullanarak $metadata üst bilgileri de girebilirsiniz. Sözdizimi yalnızca mesaj başlığının bir parçası olan MQTT özellikleri için gereklidir. Daha fazla bilgi için, bkz: alan referansları.

   Koşul, kaynak verilerdeki alanları kullanabilir. Örneğin, sıcaklık alanına göre 20'den küçük veya buna eşit verileri filtrelemek için gibi @temperature > 20 bir filtre koşulu kullanabilirsiniz.

3. Uygula’yı seçin.

Azure CLI

Örneğin, verileri filtrelemek temperature için kaynak verilerdeki alanını kullanabilirsiniz:

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

Biceps

Örneğin, verileri filtrelemek temperature için kaynak verilerdeki alanını kullanabilirsiniz:

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

temperature Alan 20'den büyükse, veriler bir sonraki aşamaya geçirilir. temperature Alan 20'den küçük veya buna eşitse, veriler filtrelenmiş olur.

Kubernetes (önizleme)

Örneğin, verileri filtrelemek temperature için kaynak verilerdeki alanını kullanabilirsiniz:

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

temperature Alan 20'den büyükse, veriler bir sonraki aşamaya geçirilir. temperature Alan 20'den küçük veya buna eşitse, veriler filtrelenmiş olur.

Haritalama: Verileri bir alandan diğerine taşımak

İsteğe bağlı dönüştürme ile verileri başka bir alana eşlemek için işlemi kullanabilirsiniz map . Dönüştürme, kaynak verilerdeki alanları kullanan bir formül olarak belirtilir.

İşlem deneyiminde eşleme şu anda İşlem, Yeniden Adlandırma ve Yeni özellik dönüşümleri kullanılarak desteklenmektedir.

İşlem

Kaynak verilere formül uygulamak için İşlem dönüşümünü kullanabilirsiniz. Bu işlem, kaynak verilere formül uygulamak ve sonuç alanını depolamak için kullanılır.

1. " Dönüştür (isteğe bağlı) altında İşlem > Ekle seçeneğini seçin."

   İşlem deneyimi kullanılarak hesaplama dönüşümü ekleme sürecine ait ekran görüntüsü.

2. Gerekli ayarları girin.

   Ayarlar	Açıklama
   Formül seçme	Açılan listeden var olan bir formülü seçin veya özel'i seçerek formülü el ile girin.
   Çıktı	Sonuç için çıktı ekran adını belirtin.
   Formül	Kaynak verilere uygulanacak formülü girin.
   Açıklama	Dönüştürme için bir açıklama sağlayın.
   Bilinen son değer	İsteğe bağlı olarak, geçerli değer kullanılamıyorsa bilinen son değeri kullanın.

   Formül alanına formül girebilir veya düzenleyebilirsiniz. Formül, kaynak verilerdeki alanları kullanabilir. Açılan listeden veri noktaları seçmek için Ctrl + Boşluk tuşlarını girin @ veya seçin. Yerleşik formüller için yer tutucuyu <dataflow> seçerek kullanılabilir veri noktalarının listesini görebilirsiniz.

   veya @$metadata.user_properties.<property>biçimini @$metadata.topic kullanarak MQTT meta veri özelliklerini girebilirsiniz. biçimini @$metadata.<header>kullanarak $metadata üst bilgileri de girebilirsiniz. Sözdizimi yalnızca mesaj başlığının bir parçası olan MQTT özellikleri için gereklidir. Daha fazla bilgi için, bkz: alan referansları.

   Formül, kaynak verilerdeki alanları kullanabilir. Örneğin, sıcaklığı Santigrat'a dönüştürmek ve çıkış alanında depolamak için kaynak verilerdeki temperature alanı kullanabilirsiniztemperatureCelsius.

3. Uygula’yı seçin.

Yeniden Adlandır

Yeniden Adlandır dönüşümünü kullanarak bir veri noktasını yeniden adlandırabilirsiniz . Bu işlem, kaynak verilerdeki bir veri noktasını yeni bir adla yeniden adlandırmak için kullanılır. Yeni ad, veri akışının sonraki aşamalarında kullanılabilir.

1. Dönüştür (isteğe bağlı) altında Yeniden Adlandır> seçin.

   

2. Gerekli ayarları girin.

   Ayarlar	Açıklama
   Veri noktası	Açılan listeden bir veri noktası seçin veya $metadata üst bilgisi girin.
   Yeni veri noktası adı	Veri noktası için yeni adı girin.
   Açıklama	Dönüştürme için bir açıklama sağlayın.

   veya @$metadata.user_properties.<property>biçimini @$metadata.topic kullanarak MQTT meta veri özelliklerini girebilirsiniz. biçimini @$metadata.<header>kullanarak $metadata üst bilgileri de girebilirsiniz. Sözdizimi yalnızca mesaj başlığının bir parçası olan MQTT özellikleri için gereklidir. Daha fazla bilgi için, bkz: alan referansları.

3. Uygula’yı seçin.

Yeni özellik

Yeni özellik dönüşümünü kullanarak kaynak verilere yeni bir özellik ekleyebilirsiniz. Bu işlem, kaynak verilere yeni bir özellik eklemek için kullanılır. Yeni özellik, veri akışının sonraki aşamalarında kullanılabilir.

İşlem deneyimi

1. Dönüştür (isteğe bağlı) altında Yeni özellik> seçin.

   

2. Gerekli ayarları girin.

   Ayarlar	Açıklama
   Özellik anahtarı	Yeni özelliğin anahtarını girin.
   Özellik değeri	Yeni özelliğin değerini girin.
   Açıklama	Yeni özellik hakkında bir açıklama sağlayın.

3. Uygula’yı seçin.

Azure CLI

Örneğin, sıcaklığı Santigrat'a dönüştürmek ve alanda depolamak için kaynak verilerdeki temperature alanı kullanabilirsiniztemperatureCelsius. Kaynak verileri bağlamsallaştırma veri kümesindeki location alanla da zenginleştirebilirsiniz:

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

Biceps

veya $metadata.user_properties.<property>biçimini $metadata.topic kullanarak MQTT meta veri özelliklerine erişebilirsiniz. biçimini $metadata.<header>kullanarak $metadata üst bilgileri de girebilirsiniz. Daha fazla bilgi için, bkz: alan referansları.

Örneğin, sıcaklığı Santigrat'a dönüştürmek ve alanda depolamak için kaynak verilerdeki temperature alanı kullanabilirsiniztemperatureCelsius. Kaynak verileri bağlamsallaştırma veri kümesindeki location alanla da zenginleştirebilirsiniz:

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

Kubernetes (önizleme)

veya $metadata.user_properties.<property>biçimini $metadata.topic kullanarak MQTT meta veri özelliklerine erişebilirsiniz. biçimini $metadata.<header>kullanarak $metadata üst bilgileri de girebilirsiniz. Daha fazla bilgi için, bkz: alan referansları.

Örneğin, sıcaklığı Santigrat'a dönüştürmek ve alanda depolamak için kaynak verilerdeki temperature alanı kullanabilirsiniztemperatureCelsius. Kaynak verileri bağlamsallaştırma veri kümesindeki location alanla da zenginleştirebilirsiniz:

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

Daha fazla bilgi edinmek için bkz. Veri akışlarını kullanarak verileri eşleme ve Veri akışlarını kullanarak verileri dönüştürme.

Kaldır

Varsayılan olarak, tüm veri noktaları çıkış şemasına eklenir. Kaldır dönüşümünü kullanarak hedeften herhangi bir veri noktasını kaldırabilirsiniz .

İşlem deneyimi

1. Dönüştür (isteğe bağlı) altındaKaldır'ı seçin.

2. Çıkış şemasından kaldırılacak veri noktasını seçin.

   

3. Uygula’yı seçin.

Azure CLI

Çıktı şemasından bir veri noktasını kaldırmak için veri akışı yapılandırmasında özelliğini kullanabilirsiniz builtInTransformationSettings . map özelliği kaldırılacak veri noktalarını belirtmek için kullanılır.

{
  "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": ""
      }
    ]
  }
}

Biceps

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

Kubernetes (önizleme)

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: ""

Daha fazla bilgi edinmek için bkz. Veri akışlarını kullanarak verileri eşleme ve Veri akışlarını kullanarak verileri dönüştürme.

Verileri şemaya göre seri hale getirme

Verileri hedefe göndermeden önce seri hale getirmek istiyorsanız, bir şema ve serileştirme biçimi belirtmeniz gerekir. Aksi takdirde, veriler JSON'da türleri türetilmiş şekilde serileştirilir. Microsoft Fabric veya Azure Data Lake gibi depolama uç noktaları, veri tutarlılığını sağlamak için bir şema gerektirir. Desteklenen serileştirme biçimleri Parquet ve Delta'dır.

İpucu

Şemayı örnek bir veri dosyasından oluşturmak için Şema Nesil Yardımcısı'nı kullanın.

İşlem deneyimi

İşlem deneyimi için, veri akışı uç noktası ayrıntılarında şema ve serileştirme biçimini belirtirsiniz. Serileştirme biçimlerini destekleyen uç noktalar Microsoft Fabric OneLake, Azure Data Lake Storage 2. Nesil, Azure Veri Gezgini ve yerel depolamadır. Örneğin, verileri Delta biçiminde seri hale getirmek için şema kayıt defterine bir şema yüklemeniz ve veri akışı hedef uç noktası yapılandırmasında buna başvurmanız gerekir.

Azure CLI

Şema kayıt defterine bir şema yükledikten sonra, buna veri akışı yapılandırmasında başvurabilirsiniz.

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

Biceps

Şema kayıt defterine bir şema yükledikten sonra, buna veri akışı yapılandırmasında başvurabilirsiniz.

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

Kubernetes (önizleme)

Şema kayıt defterine bir şema yükledikten sonra, buna veri akışı yapılandırmasında başvurabilirsiniz.

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

Şema kayıt defteri hakkında daha fazla bilgi için bkz . İleti şemalarını anlama.

Hedef

Veri akışı için bir hedef yapılandırmak için uç nokta başvuru ve veri hedefini belirtin. Uç nokta için veri hedeflerinin listesini belirtebilirsiniz.

Yerel MQTT aracısı dışındaki bir hedefe veri göndermek için bir veri akışı uç noktası oluşturun. Nasıl yapılacağını öğrenmek için bkz . Veri akışı uç noktalarını yapılandırma. Hedef yerel MQTT aracısı değilse, kaynak olarak kullanılmalıdır. Yerel MQTT aracı uç noktasını kullanma hakkında daha fazla bilgi edinmek için bkz. Veri akışları yerel MQTT aracı uç noktasını kullanmalıdır.

Önemli

Depolama uç noktaları, serileştirme için bir şema gerektirir. Microsoft Fabric OneLake, Azure Data Lake Storage, Azure Veri Gezgini veya Yerel Depolama ile veri akışını kullanmak için bir şema başvurusu belirtmeniz gerekir.

İşlem deneyimi

1. Hedef olarak kullanılacak veri akışı uç noktasını seçin.

   

   Depolama uç noktaları, serileştirme için bir şema gerektirir. Microsoft Fabric OneLake, Azure Data Lake Storage, Azure Veri Gezgini veya Yerel Depolama hedef uç noktasını seçerseniz bir şema başvurusu belirtmeniz gerekir. Örneğin, verileri Delta biçiminde bir Microsoft Fabric uç noktasına seri hale getirmek için şema kayıt defterine bir şema yüklemeniz ve veri akışı hedef uç noktası yapılandırmasında buna başvurmanız gerekir.

   

2. Hedefi yapılandırmak için Devam Et'i seçin.

3. Verilerin göndereceği konu veya tablo da dahil olmak üzere hedef için gerekli ayarları girin. Daha fazla bilgi için bkz. Veri hedefini yapılandırma (konu, kapsayıcı veya tablo).

Azure CLI

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

Biceps

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

Kubernetes (önizleme)

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

Veri hedefini yapılandırma (konu, kapsayıcı veya tablo)

Veri kaynaklarına benzer şekilde, veri hedefi de veri akışı uç noktalarını birden çok veri akışında yeniden kullanılabilir durumda tutmak için kullanılan bir kavramdır. Temelde, veri akışı uç noktası yapılandırmasındaki alt dizini temsil eder. Örneğin, veri akışı uç noktası bir depolama uç noktasıysa, veri hedefi depolama hesabındaki tablodur. Veri akışı uç noktası bir Kafka uç noktasıysa, veri hedefi Kafka konusudur.

Uç nokta türü	Veri hedefi anlamı	Açıklama
MQTT (veya Event Grid)	Konu	Verilerin gönderildiği MQTT konusu. ve ${inputTopic}gibi ${inputTopic.index} değişkenleri kullanarak hem statik konuları hem de dinamik konu çevirilerini destekler. Daha fazla bilgi için bkz . Dinamik hedef konuları.
Kafka (veya Event Hubs)	Konu	Verilerin gönderildiği Kafka başlığı. Yalnızca statik konular desteklenir, joker karakterler desteklenmez. Uç nokta bir Event Hubs ad alanıysa, veri hedefi ad alanındaki tek olay hub'ıdır.
Azure Data Lake Storage	Kapsayıcı	Depolama hesabındaki kapsayıcı. Masa değil.
Microsoft Fabric OneLake	Tablo veya Klasör	Uç nokta için yapılandırılmış yol türüne karşılık gelir.
Azure Veri Gezgini	Tablo	Azure Veri Gezgini veritabanındaki tablo.
Yerel Depolama	Klasör	Yerel depolama kalıcı birim montajındaki klasör veya dizin adı. Azure Container Storage, Azure Arc Cloud Ingest Edge Hacimleri tarafından etkinleştirildiğinde kullanılırken, bunun, oluşturduğunuz altvolum parametresiyle eşleşmesi gerekir.
OpenTelemetry	Konu	Verilerin gönderildiği OpenTelemetry konusu. Yalnızca statik konular desteklenir.

Veri hedefini yapılandırmak için:

İşlem deneyimi

İşlem deneyimi kullanılırken, veri hedefi alanı uç nokta türüne göre otomatik olarak yorumlanır. Örneğin, veri akışı uç noktası bir depolama uç noktasıysa, hedef ayrıntılar sayfası kapsayıcı adını girmenizi ister. Veri akışı uç noktası bir MQTT uç noktasıysa, hedef ayrıntılar sayfası konuyu girmenizi ister, vb.

Azure CLI

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

Örneğin, verileri yerel MQTT aracısına statik bir MQTT konusuna geri göndermek için aşağıdaki yapılandırmayı kullanın:

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

Veya özel olay hub'ı uç noktanız varsa yapılandırma şöyle görünür:

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

MQTT uç noktaları için dinamik konu değişkenlerini de kullanabilirsiniz. Örneğin, iletileri factory/1/data'den processed/factory/1'e yönlendirmek için:

{
  "destinationSettings": {
    "endpointRef": "default",
    "dataDestination": "processed/factory/${inputTopic.2}"
  }
}

Biceps

Söz dizimi tüm veri akışı uç noktaları için aynıdır:

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

Örneğin, verileri yerel MQTT aracısına statik bir MQTT konusuna geri göndermek için aşağıdaki yapılandırmayı kullanın:

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

Veya özel olay hub'ı uç noktanız varsa yapılandırma şöyle görünür:

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

Hedef olarak depolama uç noktası kullanan başka bir örnek:

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

MQTT uç noktaları için dinamik konu değişkenlerini de kullanabilirsiniz:

destinationSettings: {
  endpointRef: 'default'
  dataDestination: 'processed/factory/${inputTopic.2}'
}

Kubernetes (önizleme)

Söz dizimi tüm veri akışı uç noktaları için aynıdır:

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

Örneğin, verileri yerel MQTT aracısına statik bir MQTT konusuna geri göndermek için aşağıdaki yapılandırmayı kullanın:

destinationSettings:
  endpointRef: default
  dataDestination: example-topic

Veya özel olay hub'ı uç noktanız varsa yapılandırma şöyle görünür:

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

Hedef olarak depolama uç noktası kullanan başka bir örnek:

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

MQTT uç noktaları için dinamik konu değişkenlerini de kullanabilirsiniz:

destinationSettings:
  endpointRef: default
  dataDestination: processed/factory/${inputTopic.2}

Dinamik hedef konuları

MQTT uç noktaları için, kaynak konu yapısına göre iletileri yönlendirmek için alandaki dinamik konu değişkenlerini dataDestination kullanabilirsiniz. Aşağıdaki değişkenler kullanılabilir:

• ${inputTopic} - Tam özgün giriş konusu
• ${inputTopic.index} - Giriş konusunun bir bölümü (indeks 1'den başlar)

Örneğin, processed/factory/${inputTopic.2} iletileri 'den factory/1/data adresine processed/factory/1yönlendirir. Konu segmentleri 1'den başlayan indeksli ve baştaki/sondaki eğik çizgiler yoksayılır.

Bir konu değişkeni çözümlenemiyorsa (örneğin, ${inputTopic.5} giriş konusunun yalnızca üç kesimi olduğunda), ileti bırakılır ve bir uyarı günlüğe kaydedilir. Hedef konu başlıklarında joker karakterlere (# ve +) izin verilmez.

Not

$, { ve } karakterleri MQTT konu adlarında geçerli olduğundan, factory/$inputTopic.2 gibi bir konu kullanılabilir ancak dinamik konu değişkeni kullanmayı amaçladıysanız yanlıştır.

Örnek

Aşağıdaki örnek, kaynak ve hedef için MQTT uç noktasını kullanan bir veri akışı yapılandırmasıdır. Kaynak, MQTT konu başlığındaki azure-iot-operations/data/thermostatverileri filtreler. Dönüşüm, sıcaklığı Fahrenheit'e dönüştürür ve sıcaklığın nemle çarpıldığı verilerin 100000'den az olduğu verileri filtreler. Hedef, verileri MQTT konusuna factorygönderir.

İşlem deneyimi

Azure CLI

Veri akışı oluşturmak veya değiştirmek için az iot ops dataflow apply komutunu kullanın.

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

--config-file parametresi, kaynak özelliklerini içeren bir JSON yapılandırma dosyasının yolu ve dosya adıdır.

Bu örnekte, kullanıcının giriş dizininde depolanan aşağıdaki içeriğe sahip adlı data-flow.json bir yapılandırma dosyası varsayın:

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

Varsayılan veri akışı profilini kullanarak veri akışı oluşturmak veya güncelleştirmek için örnek bir komut aşağıda verilmişti:

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

Farklı termostatlardan gelen iletileri cihaza özgü konulara yönlendirmek için dinamik konu çevirisi kullanan başka bir örnek aşağıda verilmiştir:

{
    "mode": "Enabled",
    "operations": [
      {
        "operationType": "Source",
        "sourceSettings": {
          "dataSources": [
            "thermostats/+/sensor/temperature"
          ],
          "endpointRef": "default",
          "serializationFormat": "Json"
        }
      },
      {
        "destinationSettings": {
          "dataDestination": "processed/device/${inputTopic.2}/temperature",
          "endpointRef": "default"
        },
        "operationType": "Destination"
      }
    ]
}

Bu yapılandırma thermostats/device1/sensor/temperature iletilerini işler ve processed/device/device1/temperature gönderir.

Biceps

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

Kubernetes (önizleme)

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

Veri akışı yapılandırmalarına daha fazla örnek görmek için bkz. Azure REST API - Veri akışı ve Hızlı Başlangıç Bicep.

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

Azure Event Grid'e veri akışının çalıştığını doğrulamak için Öğretici: çift yönlü MQTT köprüsünü izleyin.

Veri akışı yapılandırmasını dışarı aktarma

Veri akışı yapılandırmasını dışarı aktarmak için işlem deneyimini veya veri akışı özel kaynağını dışarı aktararak kullanabilirsiniz.

İşlem deneyimi

Dışarı aktarmak istediğiniz veri akışını seçin ve araç çubuğundan Dışarı Aktar'ı seçin.

Azure CLI

Veri akışını dışarı aktarmak için az iot ops dataflow show komutunu kullanın.

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

Aşağıda, data-flow adlı bir veri akışını data-flow.json adlı bir JSON dosyasına dışarı aktarmak için örnek bir komut:

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

Biceps

Bicep, kod olarak altyapıdır ve dışa aktarım gerekmez. Veri akışlarını hızla ayarlamak ve yapılandırmak üzere bir veri akışı oluşturmak için Bicep dosyasını kullanın.

Kubernetes (önizleme)

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

Uygun veri akışı yapılandırması

Veri akışının beklendiği gibi çalıştığından emin olmak için aşağıdakileri doğrulayın:

• Varsayılan MQTT veri akışı uç noktası kaynak veya hedef olarak kullanılmalıdır.
• Veri akışı profili vardır ve veri akışı yapılandırmasında başvurulur.
• Kaynak bir MQTT uç noktası, Kafka uç noktası veya bir varlıktır. Depolama türü uç noktaları kaynak olarak kullanılamaz.
• Kaynak olarak Event Grid kullanılırken, Event Grid MQTT aracısı paylaşılan abonelikleri desteklemediğinden veri akışı profil örneği sayısı 1 olarak ayarlanır.
• Kaynak olarak Event Hubs kullanılırken, ad alanındaki her olay hub'ı ayrı bir Kafka konusudur ve veri kaynağı olarak belirtilmelidir.
• Dönüştürme, kullanılırsa, özel karakterlerin doğru kaçışı da dahil olmak üzere uygun söz dizimi ile yapılandırılır.
• Depolama türü uç noktaları hedef olarak kullanılırken bir şema belirtilir.
• MQTT uç noktaları için dinamik hedef konuları kullanırken konu değişkenlerinin geçerli segmentlere başvurduğundan emin olun.

Sonraki adımlar

• Veri akışlarını kullanarak verileri eşleme
• Veri akışlarını kullanarak verileri dönüştürme
• Veri akışlarını kullanarak verileri zenginleştirme
• İleti şemalarını anlama
• Veri akışı profillerini yönetme