Azure IoT İşlemleri'nde veri akışı kaynağı yapılandırma

Kaynak, verilerin bir veri akışına veya veri akışı grafiğine girdiği yerdir. Bu uç nokta için bir uç nokta başvurusu ve veri kaynaklarının (konular) listesini belirterek kaynağı yapılandırabilirsiniz.

Tip

Tek bir veri akışı kaynağı aynı anda birden çok MQTT veya Kafka konusuna abone olabilir. Her konu için ayrı veri akışları oluşturmanız gerekmez. dataSources Joker karakterler de dahil olmak üzere birden fazla konu filtresi eklemek için bu alanı (veya Konular>Satır Ekle seçeneklerini Operasyon deneyiminde) kullanın. Daha fazla bilgi için bkz. Birden çok konuya abone olma.

Bu sayfa hem veri akışları hem de veri akışı grafikleri için geçerlidir. Veri akışları için kaynak, Dataflow kaynağında bir işlemdir. Veri akışı grafları için kaynak, kaynaktaki Source bir DataflowGraph düğümdür.

Önemli

Veri akışları MQTT ve Kafka kaynak uç noktalarını destekler. Veri akışı grafikleri MQTT ve Kafka kaynak uç noktalarını destekler. Her veri akışının kaynak veya hedef olarak Azure IoT İşlemleri yerel MQTT aracısı varsayılan uç noktasına sahip olması gerekir. Daha fazla bilgi için bkz . Veri akışları yerel MQTT aracı uç noktasını kullanmalıdır.

Kaynak olarak aşağıdaki seçeneklerden birini kullanabilirsiniz.

Varsayılan uç noktayı kullanma

İşlem deneyimi

1. Kaynak ayrıntıları'nın altında Veri akışı uç noktası'yı seçin.

   

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

   Setting	Açıklama
   Veri akışı uç noktası	Varsayılan MQTT ileti aracısı uç noktasını kullanmak için varsayılan'ı seçin.
   Başlık	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. Birden çok konuya abone olma.
   İleti şeması	Gelen iletileri seri durumdan çıkarmak için kullanılacak şema. Verileri seri durumdan çıkarmak için şema belirtme konusuna bakın.

3. seçin, sonra daUygula'yı seçin.

Azure CLI

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

Biceps

sourceSettings: {
  endpointRef: 'default'
  dataSources: [
    'thermostats/+/sensor/temperature/#'
    'humidifiers/+/sensor/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.

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ı kabul ettiğinden, konular farklı olsa bile birden çok veri akışı için uç noktayı yeniden kullanabilirsiniz. Daha fazla bilgi için bkz. Birden çok konuya abone olma.

Varlığı kaynak olarak kullanma

İşlem deneyimi

Veri akışının kaynağı olarak bir varlığı kullanabilirsiniz. Bir varlığı yalnızca operasyon deneyiminde kaynak olarak kullanabilirsiniz.

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ığı yalnızca işlem deneyiminde kaynak olarak yapılandırabilirsiniz.

Biceps

Bir varlığı yalnızca işlem deneyiminde kaynak olarak yapılandırabilirsiniz.

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.

Bir varlığı yalnızca işlem deneyiminde kaynak olarak yapılandırabilirsiniz.

Kaynak olarak bir varlık kullandığınızda, varlık tanımı veri akışı için şemayı sağlar. Varlık tanımı, varlığın veri noktalarının şemasını içerir. Daha fazla bilgi için bkz. Varlık yapılandırmalarını uzaktan yönetme.

Kaynağı yapılandırdıktan sonra, varlıktan gelen veriler yerel MQTT aracısı aracılığıyla veri akışına ulaşır. Bu nedenle, bir varlığı kaynak olarak kullandığınızda, veri akışı yerel MQTT broker'ının varsayılan uç noktasını kaynak olarak kullanır.

Özel MQTT veya Kafka uç noktası 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. Veri Gölü 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 Veri akışı uç noktası'yı seçin.

   

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

   Setting	Açıklama
   Veri akışı uç noktası	Özel bir MQTT veya Kafka data flow uç noktası seçmek için Reselect düğmesini kullanın. Daha fazla bilgi için bkz. MQTT data flow uç noktalarını yapılandırma veya Azure Event Hubs ve Kafka data flow uç noktalarını yapılandırma.
   Başlık	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. Birden çok konuya abone olma.
   İleti şeması	Gelen iletileri seri durumdan çıkarmak için kullanılacak şema. Verileri seri durumdan çıkarmak için şema belirtme konusuna bakın.

3. seçin, sonra daUygula'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>'
  ]
}

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.

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>

Birden çok konuya abone olma

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ç noktayı birden çok veri akışında yeniden kullanabileceğiniz anlamına gelir. Daha fazla bilgi için bkz . Veri akışı uç noktalarını yeniden kullanma.

MQTT konu joker karakterleri

Kaynak bir MQTT (Event Grid dahil) uç noktası olduğunda, gelen iletilere abone olmak için MQTT konu filtresi kullanın. 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 Veri akışı uç noktası'nı seçin, ardından Konu(lar) alanını kullanarak gelen iletiler için abone olunacak MQTT konu filtrelerini belirtin. Birden çok MQTT konusu eklemek için Satır ekle'yi seçin ve yeni bir konu girin.

Azure CLI

{
    "operationType": "Source",
    "sourceSettings": {
        "endpointRef": "<MESSAGE_BROKER_ENDPOINT_NAME>",
        "dataSources": [
            "<TOPIC_FILTER_1>",
            "<TOPIC_FILTER_2>"
        ]
    }
}

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 karakter +, thermostats ve humidifiers konularının altındaki tüm cihazları seçer. Joker karakter, # ve temperature konularının tüm alt konuları altındaki tüm algılayıcı iletilerini seçer.

Biceps

sourceSettings: {
  endpointRef: '<MESSAGE_BROKER_ENDPOINT_NAME>'
  dataSources: [
    '<TOPIC_FILTER_1>'
    '<TOPIC_FILTER_2>'
  ]
}

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

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

Burada, joker karakter +, thermostats ve humidifiers konularının altındaki tüm cihazları seçer. Joker karakter, # ve temperature konularının tüm alt konuları altındaki tüm algılayıcı iletilerini seçer.

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.

sourceSettings:
  endpointRef: <ENDPOINT_NAME>
  dataSources:
    - <TOPIC_FILTER_1>
    - <TOPIC_FILTER_2>

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

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

Burada, joker karakter +, thermostats ve humidifiers konularının altındaki tüm cihazları seçer. Joker karakteri #, temperature ve humidity konularının tüm alt konuları altındaki tüm iletileri seçer.

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>belirtin.

İşlem deneyimi

İşlem deneyimi veri akışı Kaynak ayrıntıları bölümünde Veri akışı uç noktası'nıseçin ve Konu alanını kullanarak paylaşılan abonelik grubunu ve konusunu belirtin.

Azure CLI

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

Biceps

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

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.

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 konuyu açıkça eklemeniz önerilmez çünkü $shared ön eki gerektiğinde otomatik olarak eklenir. Veri akışları, ayarlanmadıysa grup adıyla optimizasyon yapabilir. Örneğin, ayarlı değilse $shared veri akışlarının yalnızca konu adı üzerinde çalışması gerekir.

Önemli

Paylaşılan abonelikler, örnek sayısı birden büyük olduğunda ve paylaşılan abonelikleri desteklemediğinden kaynak olarak Event Grid MQTT aracısını kullandığınızda veri akışları için ö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.

Uyarı

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 Veri akışı uç noktası'nı seçin, ardından Konu alanını kullanarak gelen iletilere abone olunacak Kafka konu filtresi belirtin.

Uyarı

İşlem deneyiminde yalnızca bir konu filtresi belirtebilirsiniz. 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>"
    ]
  }
}

Biceps

sourceSettings: {
  endpointRef: '<KAFKA_ENDPOINT_NAME>'
  dataSources: [
    '<KAFKA_TOPIC_1>'
    '<KAFKA_TOPIC_2>'
  ]
}

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.

sourceSettings:
  endpointRef: <KAFKA_ENDPOINT_NAME>
  dataSources:
    - <KAFKA_TOPIC_1>
    - <KAFKA_TOPIC_2>

Hedef yolda kaynak konuyu kullanma

Joker karakterlerle birden çok konuya abone olduğunuzda, kaynak konuyu hedef yolda değişken olarak kullanabilirsiniz. Bu özellik hem veri akışları hem de veri akışı graflarıyla çalışır.

Tam kaynak konu başlığı için veya ${inputTopic} belirli bir kesimi (1 dizinli) ayıklamak için kullanın${inputTopic.N}. Örneğin, factory/+/telemetry/# aboneliğine abone olduysanız, factory/line1/telemetry/temp üzerine gelen bir ileti, processed/${inputTopic.2}/data gibi bir hedef konuya yönlendirilebilir ve bu da processed/line1/data olarak çözümlenebilir.

Tüm ayrıntılar ve örnekler için bkz . Dinamik hedef konuları.

Kaynak şemayı belirtme

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

Kaynak bir varlıksa portal, şemayı varlık tanımından otomatik olarak çıkartır.

Tip

Şemayı örnek bir veri dosyasından oluşturmak için Schema Gen Yardımcısı 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 Veri akışı uç noktasını seçin ve şemayı belirtmek için İleti şeması alanını kullanın. Şema dosyasını karşıya yüklemek için "Karşıya Yükle"'yi seçin. Daha fazla bilgi 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şvurun.

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

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.

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

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

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

Sonraki Adımlar

• Veri akışı hedefi yapılandırma
• Veri akışı oluşturma
• Veri akışı graflarına genel bakış