Mengonfigurasi aliran data di Operasi Azure IoT

Penting

Pratinjau Operasi Azure IoT – diaktifkan oleh Azure Arc saat ini dalam pratinjau. Anda tidak boleh menggunakan perangkat lunak pratinjau ini di lingkungan produksi.

Anda harus menyebarkan penginstalan Azure IoT Operations baru saat rilis yang tersedia secara umum tersedia. Anda tidak akan dapat memutakhirkan penginstalan pratinjau.

Untuk persyaratan hukum yang berlaku untuk fitur Azure yang beta, dalam pratinjau, atau belum dirilis ke ketersediaan umum, lihat Ketentuan Penggunaan Tambahan untuk Pratinjau Microsoft Azure.

Aliran data adalah jalur yang diambil data dari sumber ke tujuan dengan transformasi opsional. Anda dapat mengonfigurasi aliran data dengan membuat sumber daya kustom Aliran Data atau menggunakan portal Azure IoT Operations Studio. Aliran data terdiri dari tiga bagian: sumber, transformasi, dan tujuan.

Untuk menentukan sumber dan tujuan, Anda perlu mengonfigurasi titik akhir aliran data. Transformasi bersifat opsional dan dapat mencakup operasi seperti memperkaya data, memfilter data, dan memetakan data ke bidang lain.

Penting

Setiap aliran data harus memiliki titik akhir default broker MQTT lokal Azure IoT Operations sebagai sumber atau tujuan.

Anda dapat menggunakan pengalaman operasi di Operasi Azure IoT untuk membuat aliran data. Pengalaman operasi menyediakan antarmuka visual untuk mengonfigurasi aliran data. Anda juga dapat menggunakan Bicep untuk membuat aliran data menggunakan file templat Bicep, atau menggunakan Kubernetes untuk membuat aliran data menggunakan file YAML.

Lanjutkan membaca untuk mempelajari cara mengonfigurasi sumber, transformasi, dan tujuan.

Prasyarat

Anda dapat menyebarkan aliran data segera setelah Anda memiliki instans Pratinjau Operasi Azure IoT menggunakan profil aliran data dan titik akhir default. Namun, Anda mungkin ingin mengonfigurasi profil aliran data dan titik akhir untuk menyesuaikan aliran data.

Profil aliran data

Profil aliran data menentukan jumlah instans untuk aliran data di bawahnya untuk digunakan. Jika Anda tidak memerlukan beberapa grup aliran data dengan pengaturan penskalaan yang berbeda, Anda dapat menggunakan profil aliran data default. Untuk mempelajari cara mengonfigurasi profil aliran data, lihat Mengonfigurasi profil aliran data.

Titik akhir aliran data

Titik akhir aliran data diperlukan untuk mengonfigurasi sumber dan tujuan untuk aliran data. Untuk memulai dengan cepat, Anda dapat menggunakan titik akhir aliran data default untuk broker MQTT lokal. Anda juga dapat membuat jenis titik akhir aliran data lainnya seperti Kafka, Event Hubs, atau Azure Data Lake Storage. Untuk mempelajari cara mengonfigurasi setiap jenis titik akhir aliran data, lihat Mengonfigurasi titik akhir aliran data.

Memulai

Setelah memiliki prasyarat, Anda dapat mulai membuat aliran data.

Portal

Untuk membuat aliran data dalam pengalaman operasi, pilih Aliran Data>Buat aliran data. Kemudian, Anda akan melihat halaman tempat Anda dapat mengonfigurasi sumber, transformasi, dan tujuan untuk aliran data.

Bicep

Buat file Bicep .bicep untuk mulai membuat aliran data. Contoh ini menunjukkan struktur aliran data yang berisi konfigurasi sumber, transformasi, dan tujuan.

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

resource aioInstance 'Microsoft.IoTOperations/instances@2024-09-15-preview' existing = {
  name: aioInstanceName
}

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

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

// Pointer to the default dataflow profile
resource defaultDataflowProfile 'Microsoft.IoTOperations/instances/dataflowProfiles@2024-09-15-preview' existing = {
  parent: aioInstance
  name: 'default'
}

resource dataflow 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflows@2024-09-15-preview' = {
  // Reference to the parent dataflow 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

Buat file manifes .yaml Kubernetes untuk mulai membuat aliran data. Contoh ini menunjukkan struktur aliran data yang berisi konfigurasi sumber, transformasi, dan tujuan.

apiVersion: connectivity.iotoperations.azure.com/v1beta1
kind: Dataflow
metadata:
  name: <DATAFLOW_NAME>
  namespace: azure-iot-operations
spec:
  # Reference to the default dataflow 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

Tinjau bagian berikut untuk mempelajari cara mengonfigurasi jenis operasi aliran data.

Sumber

Untuk mengonfigurasi sumber untuk aliran data, tentukan referensi titik akhir dan daftar sumber data untuk titik akhir.

Menggunakan aset sebagai sumber

Portal

Anda dapat menggunakan aset sebagai sumber untuk aliran data. Menggunakan aset sebagai sumber hanya tersedia dalam pengalaman operasi.

1. Di bawah Detail sumber, pilih Aset.

2. Pilih aset yang ingin Anda gunakan sebagai titik akhir sumber.

3. Pilih Lanjutkan.

   Daftar titik data untuk aset yang dipilih ditampilkan.

   

4. Pilih Terapkan untuk menggunakan aset sebagai titik akhir sumber.

Bicep

Mengonfigurasi aset sebagai sumber hanya tersedia dalam pengalaman operasi.

Kubernetes

Mengonfigurasi aset sebagai sumber hanya tersedia dalam pengalaman operasi.

Saat menggunakan aset sebagai sumber, definisi aset digunakan untuk menyimpulkan skema untuk aliran data. Definisi aset mencakup skema untuk titik data aset. Untuk mempelajari selengkapnya, lihat Mengelola konfigurasi aset dari jarak jauh.

Setelah dikonfigurasi, data dari aset mencapai aliran data melalui broker MQTT lokal. Jadi, saat menggunakan aset sebagai sumber, aliran data menggunakan titik akhir default broker MQTT lokal sebagai sumber dalam aktualitas.

Gunakan titik akhir MQTT default sebagai sumber

Portal

1. Di bawah Detail sumber, pilih MQTT.

   

2. Masukkan pengaturan berikut untuk sumber MQTT:

   Pengaturan	Deskripsi
   Topik MQTT	Filter topik MQTT untuk berlangganan pesan masuk. Lihat Mengonfigurasi topik MQTT atau Kafka.
   Skema pesan	Skema yang digunakan untuk mendeserialisasi pesan masuk. Lihat Menentukan skema untuk mendeserialisasi data.

3. Pilih Terapkan.

Bicep

Titik akhir MQTT dikonfigurasi dalam file templat Bicep. Misalnya, titik akhir berikut adalah sumber untuk aliran data.

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

Di sini, dataSources memungkinkan Anda menentukan beberapa topik MQTT atau Kafka tanpa perlu memodifikasi konfigurasi titik akhir. Fleksibilitas ini berarti titik akhir yang sama dapat digunakan kembali di beberapa aliran data, bahkan jika topik bervariasi. Untuk mempelajari selengkapnya, lihat Mengonfigurasi sumber data.

Kubernetes

Misalnya, untuk mengonfigurasi sumber menggunakan titik akhir MQTT dan dua filter topik MQTT, gunakan konfigurasi berikut:

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

Karena dataSources memungkinkan Anda menentukan topik MQTT atau Kafka tanpa memodifikasi konfigurasi titik akhir, Anda dapat menggunakan kembali titik akhir untuk beberapa aliran data meskipun topiknya berbeda. Untuk mempelajari selengkapnya, lihat Mengonfigurasi sumber data.

Jika titik akhir default tidak digunakan sebagai sumber, titik akhir harus digunakan sebagai tujuan. Untuk mempelajari selengkapnya, lihat Aliran data harus menggunakan titik akhir broker MQTT lokal.

Menggunakan titik akhir aliran data MQTT atau Kafka kustom sebagai sumber

Jika Anda membuat titik akhir aliran data MQTT atau Kafka kustom (misalnya, untuk digunakan dengan Event Grid atau Event Hubs), Anda dapat menggunakannya sebagai sumber untuk aliran data. Ingat bahwa titik akhir jenis penyimpanan, seperti Data Lake atau Fabric OneLake, tidak dapat digunakan sebagai sumber.

Untuk mengonfigurasi, gunakan Kubernetes YAML atau Bicep. Ganti nilai tempat penampung dengan nama dan topik titik akhir kustom Anda.

Portal

Menggunakan titik akhir MQTT atau Kafka kustom sebagai sumber saat ini tidak didukung dalam pengalaman operasi.

Bicep

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

Kubernetes

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

Mengonfigurasi sumber data (topik MQTT atau Kafka)

Anda dapat menentukan beberapa topik MQTT atau Kafka dalam sumber tanpa perlu mengubah konfigurasi titik akhir aliran data. Fleksibilitas ini berarti titik akhir yang sama dapat digunakan kembali di beberapa aliran data, bahkan jika topik bervariasi. Untuk informasi selengkapnya, lihat Menggunakan kembali titik akhir aliran data.

Topik MQTT

Saat sumber adalah titik akhir MQTT (Disertakan Event Grid), Anda dapat menggunakan filter topik MQTT untuk berlangganan pesan masuk. Filter topik dapat menyertakan kartubebas untuk berlangganan beberapa topik. Misalnya, thermostats/+/telemetry/temperature/# berlangganan semua pesan telemetri suhu dari termostat. Untuk mengonfigurasi filter topik MQTT:

Portal

Dalam operasi mengalami detail Sumber aliran data, pilih MQTT, lalu gunakan bidang topik MQTT untuk menentukan filter topik MQTT untuk berlangganan pesan masuk.

Catatan

Hanya satu filter topik MQTT yang dapat ditentukan dalam pengalaman operasi. Untuk menggunakan beberapa filter topik MQTT, gunakan Bicep atau Kubernetes.

Bicep

sourceSettings: {
  endpointRef: '<MQTT_ENDPOINT_NAME>'
  dataSources: [
    '<MQTT_TOPIC_FILTER_1>'
    '<MQTT_TOPIC_FILTER_2>'
    // Add more MQTT topic filters as needed
  ]
}

Contoh dengan beberapa filter topik MQTT dengan kartubebas:

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

Di sini, kartubebas + digunakan untuk memilih semua perangkat di thermostats bawah topik dan humidifiers . Kartubebas # temperature digunakan untuk memilih semua pesan telemetri di bawah semua subtopik topik dan humidity .

Kubernetes

sourceSettings:
  endpointRef: <MQTT_ENDPOINT_NAME>
  dataSources:
    - <MQTT_TOPIC_FILTER_1>
    - <MQTT_TOPIC_FILTER_2>
    # Add more MQTT topic filters as needed

Contoh dengan beberapa filter topik MQTT dengan kartubebas:

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

Di sini, kartubebas + digunakan untuk memilih semua perangkat di thermostats bawah topik dan humidifiers . Kartubebas # temperature digunakan untuk memilih semua pesan telemetri di bawah semua subtopik topik dan humidity .

Langganan bersama

Untuk menggunakan langganan bersama dengan sumber MQTT, Anda dapat menentukan topik langganan bersama dalam bentuk $shared/<GROUP_NAME>/<TOPIC_FILTER>.

Portal

Dalam operasi mengalami detail Sumber aliran data, pilih MQTT dan gunakan bidang topik MQTT untuk menentukan grup langganan dan topik bersama.

Bicep

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

Kubernetes

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

Jika jumlah instans di profil aliran data lebih besar dari 1, langganan bersama secara otomatis diaktifkan untuk semua aliran data yang menggunakan sumber MQTT. Dalam hal ini, awalan $shared ditambahkan dan nama grup langganan bersama dibuat secara otomatis. Misalnya, jika Anda memiliki profil aliran data dengan jumlah instans 3, dan aliran data Anda menggunakan titik akhir MQTT sebagai sumber yang dikonfigurasi dengan topik topic1 dan topic2, mereka secara otomatis dikonversi ke langganan bersama sebagai $shared/<GENERATED_GROUP_NAME>/topic1 dan $shared/<GENERATED_GROUP_NAME>/topic2. Jika Anda ingin menggunakan ID grup langganan bersama yang berbeda, Anda dapat mengambil alihnya dalam topik, seperti $shared/mygroup/topic1.

Penting

Aliran data yang memerlukan langganan bersama saat jumlah instans lebih besar dari 1 penting saat menggunakan broker Azure Event Grid MQTT sebagai sumber karena tidak mendukung langganan bersama. Untuk menghindari pesan yang hilang, atur jumlah instans profil aliran data ke 1 saat menggunakan broker MQTT Event Grid sebagai sumbernya. Yaitu ketika aliran data adalah pelanggan dan menerima pesan dari cloud.

Topi Kafka

Ketika sumbernya adalah titik akhir Kafka (Azure Event Hubs included), tentukan topik kafka individual untuk berlangganan pesan masuk. Kartubebas tidak didukung, jadi Anda harus menentukan setiap topik secara statis.

Catatan

Saat menggunakan Azure Event Hubs melalui titik akhir Kafka, setiap hub peristiwa individual dalam namespace adalah topik Kafka. Misalnya, jika Anda memiliki namespace Layanan Pusat Aktivitas dengan dua hub peristiwa, thermostats dan humidifiers, Anda dapat menentukan setiap hub peristiwa sebagai topik Kafka.

Untuk mengonfigurasi topik Kafka:

Portal

Menggunakan titik akhir Kafka sebagai sumber saat ini tidak didukung dalam pengalaman operasi.

Bicep

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

Kubernetes

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

Tentukan skema untuk mendeserialisasi data

Jika data sumber memiliki bidang atau bidang opsional dengan jenis yang berbeda, tentukan skema deserialisasi untuk memastikan konsistensi. Misalnya, data mungkin memiliki bidang yang tidak ada di semua pesan. Tanpa skema, transformasi tidak dapat menangani bidang ini karena akan memiliki nilai kosong. Dengan skema, Anda dapat menentukan nilai default atau mengabaikan bidang.

Menentukan skema hanya relevan saat menggunakan sumber MQTT atau Kafka. Jika sumber adalah aset, skema secara otomatis disimpulkan dari definisi aset.

Untuk mengonfigurasi skema yang digunakan untuk mendeserialisasi pesan masuk dari sumber:

Portal

Dalam operasi mengalami detail Sumber aliran data, pilih MQTT dan gunakan bidang Skema pesan untuk menentukan skema. Anda dapat menggunakan tombol Unggah untuk mengunggah file skema terlebih dahulu. Untuk mempelajari selengkapnya, lihat Memahami skema pesan.

Bicep

Setelah Anda menggunakan registri skema untuk menyimpan skema, Anda dapat mereferensikannya dalam konfigurasi aliran data.

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

Kubernetes

Setelah Anda menggunakan registri skema untuk menyimpan skema, Anda dapat mereferensikannya dalam konfigurasi aliran data.

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

Transformasi

Operasi transformasi adalah tempat Anda dapat mengubah data dari sumber sebelum mengirimkannya ke tujuan. Transformasi bersifat opsional. Jika Anda tidak perlu membuat perubahan pada data, jangan sertakan operasi transformasi dalam konfigurasi aliran data. Beberapa transformasi dirantai bersama-sama secara bertahap terlepas dari urutan yang ditentukan dalam konfigurasi. Urutan tahapan selalu:

1. Memperkaya, Mengganti Nama, atau menambahkan properti Baru: Tambahkan data tambahan ke data sumber yang diberi himpunan data dan kondisi yang cocok.
2. Filter: Memfilter data berdasarkan kondisi.
3. Peta atau Komputasi: Memindahkan data dari satu bidang ke bidang lainnya dengan konversi opsional.

Portal

Dalam pengalaman operasi, pilih Aliran Data>Tambahkan transformasi (opsional).

Bicep

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

Kubernetes

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

Pengayaan: Menambahkan data referensi

Untuk memperkaya data, Anda dapat menggunakan himpunan data referensi di penyimpanan status terdistribusi (DSS) Azure IoT Operations. Himpunan data digunakan untuk menambahkan data tambahan ke data sumber berdasarkan kondisi. Kondisi ditentukan sebagai bidang dalam data sumber yang cocok dengan bidang dalam himpunan data.

Anda dapat memuat data sampel ke DSS dengan menggunakan sampel alat set DSS. Nama kunci di penyimpanan status terdistribusi sesuai dengan himpunan data dalam konfigurasi aliran data.

Portal

Dalam pengalaman operasi, tahap Pengayaan saat ini didukung menggunakan transformasi properti Ganti Nama dan Baru.

1. Dalam pengalaman operasi, pilih aliran data lalu Tambahkan transformasi (opsional).

2. Pilih Ganti Nama atau Transformasi properti baru lalu pilih Tambahkan.

   

Bicep

Contoh ini memperlihatkan bagaimana Anda dapat menggunakan deviceId bidang dalam data sumber untuk mencocokkan asset bidang dalam himpunan data:

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

Kubernetes

Misalnya, Anda dapat menggunakan deviceId bidang dalam data sumber untuk mencocokkan asset bidang dalam himpunan data:

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

Jika himpunan data memiliki rekaman dengan asset bidang , mirip dengan:

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

Data dari sumber dengan pencocokan deviceId thermostat1 bidang memiliki location bidang dan manufacturer yang tersedia dalam tahap filter dan peta.

Untuk informasi selengkapnya tentang sintaks kondisi, lihat Memperkaya data dengan menggunakan aliran data dan Mengonversi data menggunakan aliran data.

Filter: Memfilter data berdasarkan kondisi

Untuk memfilter data pada kondisi, Anda dapat menggunakan tahapan filter . Kondisi ditentukan sebagai bidang dalam data sumber yang cocok dengan nilai.

Portal

1. Di bawah Transformasi (opsional), pilih Filter>Tambahkan.

2. Pilih titik data untuk disertakan dalam himpunan data.

3. Tambahkan kondisi dan deskripsi filter.

   

4. Pilih Terapkan.

Misalnya, Anda dapat menggunakan kondisi filter seperti temperature > 20 memfilter data kurang dari atau sama dengan 20 berdasarkan bidang suhu.

Bicep

Misalnya, Anda dapat menggunakan temperature bidang dalam data sumber untuk memfilter data:

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

temperature Jika bidang lebih besar dari 20, data diteruskan ke tahap berikutnya. temperature Jika bidang kurang dari atau sama dengan 20, data difilter.

Kubernetes

Misalnya, Anda dapat menggunakan temperature bidang dalam data sumber untuk memfilter data:

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

temperature Jika bidang lebih besar dari 20, data diteruskan ke tahap berikutnya. temperature Jika bidang kurang dari atau sama dengan 20, data difilter.

Peta: Memindahkan data dari satu bidang ke bidang lainnya

Untuk memetakan data ke bidang lain dengan konversi opsional, Anda dapat menggunakan operasi.map Konversi ditentukan sebagai rumus yang menggunakan bidang dalam data sumber.

Portal

Dalam pengalaman operasi, pemetaan saat ini didukung menggunakan transformasi Komputasi .

1. Di bawah Transformasi (opsional), pilih Tambahkan Komputasi>.

2. Masukkan bidang dan ekspresi yang diperlukan.

   

3. Pilih Terapkan.

Bicep

Misalnya, Anda dapat menggunakan temperature bidang dalam data sumber untuk mengonversi suhu ke Celcius dan menyimpannya di temperatureCelsius bidang . Anda juga dapat memperkaya data sumber dengan location bidang dari himpunan data kontekstualisasi:

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

Kubernetes

Misalnya, Anda dapat menggunakan temperature bidang dalam data sumber untuk mengonversi suhu ke Celcius dan menyimpannya di temperatureCelsius bidang . Anda juga dapat memperkaya data sumber dengan location bidang dari himpunan data kontekstualisasi:

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

Untuk mempelajari selengkapnya, lihat Memetakan data dengan menggunakan aliran data dan Mengonversi data dengan menggunakan aliran data.

Menserialisasikan data sesuai dengan skema

Jika Anda ingin membuat serialisasi data sebelum mengirimkannya ke tujuan, Anda perlu menentukan format skema dan serialisasi. Jika tidak, data diserialisasikan di JSON dengan jenis yang disimpulkan. Titik akhir penyimpanan seperti Microsoft Fabric atau Azure Data Lake memerlukan skema untuk memastikan konsistensi data. Format serialisasi yang didukung adalah Parquet dan Delta.

Portal

Saat ini, menentukan skema output dan serialisasi tidak didukung dalam pengalaman operasi.

Bicep

Setelah mengunggah skema ke registri skema, Anda dapat mereferensikannya dalam konfigurasi aliran data.

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

Kubernetes

Setelah mengunggah skema ke registri skema, Anda dapat mereferensikannya dalam konfigurasi aliran data.

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

Untuk informasi selengkapnya tentang registri skema, lihat Memahami skema pesan.

Tujuan

Untuk mengonfigurasi tujuan untuk aliran data, tentukan referensi titik akhir dan tujuan data. Anda dapat menentukan daftar tujuan data untuk titik akhir.

Untuk mengirim data ke tujuan selain broker MQTT lokal, buat titik akhir aliran data. Untuk mempelajari caranya, lihat Mengonfigurasi titik akhir aliran data. Jika tujuan bukan broker MQTT lokal, tujuan harus digunakan sebagai sumber. Untuk mempelajari selengkapnya, lihat Aliran data harus menggunakan titik akhir broker MQTT lokal.

Penting

Titik akhir penyimpanan memerlukan referensi skema. Jika Anda telah membuat titik akhir tujuan penyimpanan untuk Microsoft Fabric OneLake, ADLS Gen 2, Azure Data Explorer, dan Penyimpanan Lokal, Anda harus menentukan referensi skema.

Portal

1. Pilih titik akhir aliran data untuk digunakan sebagai tujuan.

   

2. Pilih Lanjutkan untuk mengonfigurasi tujuan.

3. Masukkan pengaturan yang diperlukan untuk tujuan, termasuk topik atau tabel untuk mengirim data. Lihat Mengonfigurasi tujuan data (topik, kontainer, atau tabel) untuk informasi selengkapnya.

Bicep

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

Kubernetes

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

Mengonfigurasi tujuan data (topik, kontainer, atau tabel)

Mirip dengan sumber data, tujuan data adalah konsep yang digunakan untuk menjaga titik akhir aliran data dapat digunakan kembali di beberapa aliran data. Pada dasarnya, ini mewakili subdirektori dalam konfigurasi titik akhir aliran data. Misalnya, jika titik akhir aliran data adalah titik akhir penyimpanan, tujuan data adalah tabel di akun penyimpanan. Jika titik akhir aliran data adalah titik akhir Kafka, tujuan data adalah topik Kafka.

Jenis titik akhir	Arti tujuan data	Deskripsi
MQTT (atau Event Grid)	Topik	Topik MQTT tempat data dikirim. Hanya topik statis yang didukung, tidak ada kartubebas.
Kafka (atau Event Hubs)	Topik	Topik Kafka tempat data dikirim. Hanya topik statis yang didukung, tidak ada kartubebas. Jika titik akhir adalah namespace Layanan Pusat Aktivitas, tujuan data adalah hub peristiwa individual dalam namespace layanan.
Azure Data Lake Storage	Kontainer	Kontainer di akun penyimpanan. Bukan meja.
Microsoft Fabric OneLake	Tabel atau Folder	Sesuai dengan jenis jalur yang dikonfigurasi untuk titik akhir.
Azure Data Explorer	Tabel	Tabel dalam database Azure Data Explorer.
Penyimpanan Lokal	Folder	Nama folder atau direktori di pemasangan volume persisten penyimpanan lokal. Saat menggunakan Azure Container Storage yang diaktifkan oleh Azure Arc Cloud Ingest Edge Volumes, ini harus cocok dengan spec.path parameter untuk subvolume yang Anda buat.

Untuk mengonfigurasi tujuan data:

Portal

Saat menggunakan pengalaman operasi, bidang tujuan data secara otomatis ditafsirkan berdasarkan jenis titik akhir. Misalnya, jika titik akhir aliran data adalah titik akhir penyimpanan, halaman detail tujuan meminta Anda untuk memasukkan nama kontainer. Jika titik akhir aliran data adalah titik akhir MQTT, halaman detail tujuan meminta Anda untuk memasukkan topik, dan sebagainya.

Bicep

Sintaksnya sama untuk semua titik akhir aliran data:

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

Misalnya, untuk mengirim data kembali ke broker MQTT lokal topik MQTT statis, gunakan konfigurasi berikut:

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

Atau, jika Anda memiliki titik akhir pusat aktivitas kustom, konfigurasinya akan terlihat seperti:

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

Contoh lain menggunakan titik akhir penyimpanan sebagai tujuan:

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

Kubernetes

Sintaksnya sama untuk semua titik akhir aliran data:

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

Misalnya, untuk mengirim data kembali ke broker MQTT lokal topik MQTT statis, gunakan konfigurasi berikut:

destinationSettings:
  endpointRef: default
  dataDestination: example-topic

Atau, jika Anda memiliki titik akhir pusat aktivitas kustom, konfigurasinya akan terlihat seperti:

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

Contoh lain menggunakan titik akhir penyimpanan sebagai tujuan:

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

Contoh

Contoh berikut adalah konfigurasi aliran data yang menggunakan titik akhir MQTT untuk sumber dan tujuan. Sumber memfilter data dari topik azure-iot-operations/data/thermostatMQTT . Transformasi mengonversi suhu ke Fahrenheit dan memfilter data di mana suhu dikalikan dengan kelembaban kurang dari 100000. Tujuan mengirimkan data ke topik factoryMQTT .

Portal

Lihat tab Bicep atau Kubernetes untuk contoh konfigurasi.

Bicep

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

resource aioInstance 'Microsoft.IoTOperations/instances@2024-09-15-preview' existing = {
  name: aioInstanceName
}

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

// Pointer to the default dataflow endpoint
resource defaultDataflowEndpoint 'Microsoft.IoTOperations/instances/dataflowEndpoints@2024-09-15-preview' existing = {
  parent: aioInstance
  name: 'default'
}

// Pointer to the default dataflow profile
resource defaultDataflowProfile 'Microsoft.IoTOperations/instances/dataflowProfiles@2024-09-15-preview' existing = {
  parent: aioInstance
  name: 'default'
}

resource dataflow 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflows@2024-09-15-preview' = {
  // Reference to the parent dataflow 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

apiVersion: connectivity.iotoperations.azure.com/v1beta1
kind: Dataflow
metadata:
  name: my-dataflow
  namespace: azure-iot-operations
spec:
  # Reference to the default dataflow 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

Untuk melihat contoh konfigurasi aliran data lainnya, lihat Azure REST API - Aliran Data dan mulai cepat Bicep.

Memverifikasi aliran data berfungsi

Ikuti Tutorial: Jembatan MQTT dua arah ke Azure Event Grid untuk memverifikasi aliran data berfungsi.

Mengekspor konfigurasi aliran data

Untuk mengekspor konfigurasi aliran data, Anda dapat menggunakan pengalaman operasi atau dengan mengekspor sumber daya kustom Aliran Data.

Portal

Pilih aliran data yang ingin Anda ekspor dan pilih Ekspor dari toolbar.

Bicep

Bicep adalah infrastruktur sebagai kode dan tidak diperlukan ekspor. Gunakan file templat Bicep untuk membuat aliran data untuk menyiapkan dan mengonfigurasi aliran data dengan cepat.

Kubernetes

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

Konfigurasi aliran data yang tepat

Untuk memastikan aliran data berfungsi seperti yang diharapkan, verifikasi hal berikut:

• Titik akhir aliran data MQTT default harus digunakan sebagai sumber atau tujuan.
• Profil aliran data ada dan dirujuk dalam konfigurasi aliran data.
• Sumber adalah titik akhir MQTT, titik akhir Kafka, atau aset. Titik akhir jenis penyimpanan tidak dapat digunakan sebagai sumber.
• Saat menggunakan Event Grid sebagai sumbernya, jumlah instans profil aliran data diatur ke 1 karena broker Azure Event Grid MQTT tidak mendukung langganan bersama.
• Saat menggunakan Azure Event Hubs sebagai sumbernya, setiap hub peristiwa di namespace adalah topik Kafka terpisah dan harus ditentukan sebagai sumber data.
• Transformasi, jika digunakan, dikonfigurasi dengan sintaks yang tepat, termasuk pelepasan karakter khusus yang tepat.
• Saat menggunakan titik akhir jenis penyimpanan sebagai tujuan, skema ditentukan.

Langkah berikutnya

• Memetakan data dengan menggunakan aliran data
• Mengonversi data dengan menggunakan aliran data
• Memperkaya data dengan menggunakan aliran data
• Memahami skema pesan
• Mengelola profil aliran data