Membuat aliran data di Operasi Azure IoT

2025-06-20

Penting

Halaman ini mencakup instruksi untuk mengelola komponen Azure IoT Operations menggunakan manifes penyebaran Kubernetes, yang saat ini dalam pratinjau. Fitur ini disediakan dengan beberapa batasan, dan tidak boleh digunakan untuk beban kerja produksi.

Lihat Ketentuan Penggunaan Tambahan untuk Pratinjau Microsoft Azure untuk persyaratan hukum yang berlaku pada fitur Azure dalam versi beta, pratinjau, atau belum dirilis secara umum.

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 antarmuka pengguna web pengalaman operasi. Aliran data terdiri dari tiga bagian: sumber, transformasi, dan tujuan.

Diagram aliran data menunjukkan aliran dari sumber ke transformasi lalu ke 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 entah 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 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 Operasi Azure IoT menggunakan profil aliran data default dan titik akhir default. Namun, Anda mungkin ingin mengonfigurasi profil aliran data dan titik akhir untuk menyesuaikan aliran data.

Profil aliran data

Jika Anda tidak memerlukan pengaturan penskalaan yang berbeda untuk aliran data Anda, gunakan profil aliran data default yang disediakan oleh Operasi Azure IoT. Anda harus menghindari mengaitkan terlalu banyak aliran data dengan satu profil aliran data. Jika Anda memiliki sejumlah besar aliran data, distribusikan di beberapa profil aliran data untuk mengurangi risiko melebihi batas ukuran konfigurasi profil aliran data.

Untuk mempelajari cara mengonfigurasi profil aliran data baru, 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.

Mulai sekarang

Setelah memiliki prasyarat, Anda dapat mulai membuat aliran data.

Untuk membuat aliran data dalam operasi pengalaman, pilih Aliran data>Buat aliran data.
Pilih nama tempat penampung new-data-flow untuk mengatur properti aliran data. Masukkan nama aliran data dan pilih profil aliran data yang akan digunakan. Profil aliran data default dipilih secara default. Untuk informasi selengkapnya tentang profil aliran data, lihat Mengonfigurasi profil aliran data.

Penting

Anda hanya dapat memilih profil aliran data saat membuat aliran data. Anda tidak dapat mengubah profil aliran data setelah aliran data dibuat. Jika Anda ingin mengubah profil aliran data dari aliran data yang ada, hapus aliran data asli dan buat yang baru dengan profil aliran data baru.
Konfigurasikan titik akhir sumber, transformasi, dan tujuan untuk aliran data dengan memilih item dalam diagram aliran data.

Gunakan perintah az iot ops dataflow apply untuk membuat atau mengubah aliran data.

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

Parameter --config-file adalah jalur dan nama file file konfigurasi JSON yang berisi properti sumber daya.

Dalam contoh ini, asumsikan file konfigurasi bernama data-flow.json dengan konten berikut yang disimpan di direktori beranda pengguna:

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

Berikut adalah contoh perintah untuk membuat atau memperbarui aliran data menggunakan profil aliran data default:

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

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

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/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

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. Pilih salah satu opsi berikut sebagai sumber untuk aliran data.

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

Opsi 1: Gunakan titik akhir broker pesan default sebagai sumber

Di bawah Detail sumber, pilih Pengelola Pesan.

Masukkan pengaturan berikut untuk sumber broker pesan:

Pengaturan	Deskripsi
Titik akhir aliran data	Pilih default untuk menggunakan titik akhir broker pesan MQTT default.
Topik	Filter topik untuk berlangganan pesan masuk. Gunakan Topik> Tambahkan baris untuk menambahkan beberapa topik. Untuk informasi selengkapnya tentang topik, lihat Mengonfigurasi topik MQTT atau Kafka.
Skema pesan	Skema yang digunakan untuk mendeserialisasi pesan masuk. Lihat Menentukan skema untuk mendeserialisasi data.

Pilih Terapkan.

Berikut adalah contoh konfigurasi titik akhir sumber untuk titik akhir broker MQTT default:

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

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

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

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

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

Opsi 2: Gunakan aset sebagai sumber

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

Di bawah Detail sumber, pilih Aset.
Pilih aset yang ingin Anda gunakan sebagai titik akhir sumber.
Pilih Lanjutkan.

Daftar titik data untuk aset yang dipilih ditampilkan.
Pilih Terapkan untuk menggunakan aset sebagai titik akhir sumber.

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.

Opsi 3: Gunakan 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.

Di bawah Detail sumber, pilih Pengelola Pesan.

Masukkan pengaturan berikut untuk sumber broker pesan:

Pengaturan	Deskripsi
Titik akhir aliran data	Gunakan tombol Pilih kembali untuk memilih titik akhir aliran data MQTT atau Kafka kustom. Untuk informasi selengkapnya, lihat Mengonfigurasi titik akhir aliran data MQTT atau Mengonfigurasi Azure Event Hubs dan titik akhir aliran data Kafka.
Topik	Filter topik untuk berlangganan pesan masuk. Gunakan Topik> Tambahkan baris untuk menambahkan beberapa topik. Untuk informasi selengkapnya tentang topik, lihat Mengonfigurasi topik MQTT atau Kafka.
Skema pesan	Skema yang digunakan untuk mendeserialisasi pesan masuk. Lihat Menentukan skema untuk mendeserialisasi data.

Pilih Terapkan.

Ganti placeholder dengan nama endpoint kustom dan topik Anda.

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

Ganti placeholder dengan nama endpoint kustom dan topik Anda.

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

Ganti placeholder dengan nama endpoint kustom dan topik Anda.

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 memodifikasi 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 sumbernya adalah titik akhir MQTT, termasuk Event Grid, Anda dapat menggunakan filter topik MQTT guna berlangganan pesan masuk. Filter topik dapat menyertakan karakter pengganti untuk berlangganan beberapa topik. Misalnya, thermostats/+/sensor/temperature/# berlangganan semua pesan sensor suhu dari termostat. Untuk mengonfigurasi filter topik MQTT:

Dalam aliran data pengalaman operasi Detail sumber, pilih Perantara Pesan, lalu gunakan kolom Topik untuk menentukan filter topik MQTT untuk berlangganan pesan yang masuk. Anda dapat menambahkan beberapa topik MQTT dengan memilih Tambahkan baris dan memasukkan topik baru.

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

Contoh dengan beberapa filter topik MQTT dengan wildcard.

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

Di sini, tanda bintang + digunakan untuk memilih semua perangkat dalam topik thermostats dan humidifiers. Kartubebas # digunakan untuk memilih semua pesan sensor di bawah semua subtopik dari topik temperature dan humidity.

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

Contoh dengan beberapa filter topik MQTT dengan wildcard.

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

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

Contoh dengan beberapa filter topik dengan pencocokan bebas:

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

Di sini, tanda bintang + digunakan untuk memilih semua perangkat dalam topik thermostats dan humidifiers. Kartubebas # digunakan untuk memilih semua pesan di bawah semua subtopik dari topik temperature dan humidity.

Langganan bersama

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

Dalam aliran data pengalaman operasi Detail sumber, pilih Broker pesan dan gunakan bidang Topik untuk menentukan grup langganan bersama dan topik.

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

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

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

Jika jumlah instans di profil aliran data lebih besar dari satu, langganan bersama secara otomatis diaktifkan untuk semua aliran data yang menggunakan sumber perantara pesan. Dalam hal ini, awalan $shared ditambahkan dan nama grup langganan bersama dibuat secara otomatis. Misalnya, jika Anda memiliki profil aliran data dengan jumlah instans sebanyak 3, dan aliran data Anda menggunakan titik akhir broker pesan 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.

Anda dapat secara eksplisit membuat topik bernama $shared/mygroup/topic dalam konfigurasi Anda. Namun, menambahkan $shared topik secara eksplisit tidak disarankan karena awalan $shared ditambahkan secara otomatis saat diperlukan. Aliran data dapat melakukan pengoptimalan dengan nama grup jika belum ditetapkan. Misalnya, $share tidak diatur dan aliran data hanya harus beroperasi melalui nama topik.

Penting

Aliran data yang memerlukan langganan bersama ketika jumlah instans lebih dari satu sangat penting saat menggunakan Azure Event Grid MQTT broker sebagai sumber karena tidak mendukung langganan bersama. Untuk menghindari pesan yang hilang, atur jumlah instans profil aliran data menjadi satu saat menggunakan broker Azure Event Grid MQTT sebagai sumbernya. Ini berarti ketika aliran data berperan sebagai pelanggan dan menerima pesan dari cloud.

Topik Kafka

Ketika sumbernya adalah titik akhir Kafka (termasuk Azure Event Hubs), tentukan topik Kafka individu untuk berlangganan pesan masuk. Wildcard tidak didukung, jadi Anda harus menentukan setiap topik dengan cara statis.

Catatan

Saat menggunakan Event Hubs melalui titik akhir Kafka, setiap hub peristiwa individual pada 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:

Dalam pengalaman operasi aliran data detail sumber, pilih broker pesan, lalu gunakan bidang topik untuk menentukan filter topik Kafka untuk berlangganan pesan masuk.

Catatan

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

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

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

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

Tentukan skema sumber

Saat menggunakan MQTT atau Kafka sebagai sumbernya, Anda dapat menentukan skema untuk menampilkan daftar titik data dalam antarmuka pengguna web pengalaman operasi. Menggunakan skema untuk mendeserialisasi dan memvalidasi pesan masuk saat ini tidak didukung.

Jika sumber adalah aset, skema secara otomatis disimpulkan dari definisi aset.

Tips

Untuk menghasilkan skema dari file data sampel, gunakan Schema Gen Helper.

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

Dalam pengalaman operasi aliran data Detail sumber, pilih Broker pesan 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.

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

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

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

Untuk mempelajari selengkapnya, lihat Memahami skema pesan.

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:

Pengayaan: Tambahkan data tambahan ke data sumber yang diberi himpunan data dan kondisi yang cocok.
Filter: Memfilter data berdasarkan kondisi.
Peta, Komputasi, Ganti Nama, atau tambahkan properti Baru: Pindahkan data dari satu bidang ke bidang lain dengan konversi opsional.

Bagian ini adalah pengenalan transformasi aliran data. Untuk informasi selengkapnya, lihat Memetakan data dengan menggunakan aliran data, Mengonversi data dengan menggunakan konversi aliran data, dan Memperkaya data dengan menggunakan aliran data.

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

Cuplikan layar antarmuka pengalaman operasi memperlihatkan penambahan tahap transformasi ke aliran data.

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

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

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

Pengayaan: Menambahkan data referensi

Untuk memperkaya data, pertama-tama tambahkan himpunan data referensi di penyimpanan keadaan Operasi Azure IoT. 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 penyimpanan status dengan menggunakan CLI penyimpanan status. Nama kunci di penyimpanan status sesuai dengan himpunan data dalam konfigurasi aliran data.

Saat ini, tahap Pengayaan tidak didukung dalam pengalaman operasi.

Untuk memperkaya data, Anda dapat menggunakan builtInTransformationSettings properti dalam konfigurasi aliran data. Properti datasets digunakan untuk menentukan himpunan data yang akan digunakan untuk pengayaan.

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

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

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 mempunyai catatan pada bidang asset, mirip dengan:

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

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

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.

Di bawah Transformasi (opsional), pilih Filter>Tambahkan.
Masukkan pengaturan yang diperlukan.

Pengaturan Deskripsi

Kondisi filter Kondisi untuk memfilter data berdasarkan bidang dalam data sumber.

Deskripsi Berikan deskripsi untuk kondisi filter.

Di bidang kondisi filter, masukkan @ atau pilih Ctrl + Spasi untuk memilih titik data dari menu dropdown.

Anda dapat memasukkan properti metadata MQTT menggunakan format @$metadata.user_properties.<property> atau @$metadata.topic. Anda juga dapat memasukkan header $metadata menggunakan format @$metadata.<header>. Sintaks $metadata hanya diperlukan untuk properti MQTT yang merupakan bagian dari header pesan. Untuk informasi selengkapnya, lihat referensi bidang.

Kondisi dapat memanfaatkan kolom dalam data sumber. Misalnya, Anda dapat menggunakan kondisi filter seperti @temperature > 20 memfilter data kurang dari atau sama dengan 20 berdasarkan bidang suhu.
Pilih Terapkan.

Pengaturan	Deskripsi
Kondisi filter	Kondisi untuk memfilter data berdasarkan bidang dalam data sumber.
Deskripsi	Berikan deskripsi untuk kondisi filter.

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

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

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.

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.

Dalam pengalaman operasi, pemetaan saat ini didukung menggunakan Komputasi, Ganti Nama, dan properti baru transformasi.

Komputasi

Anda bisa menggunakan transformasi Komputasi untuk menerapkan rumus ke data sumber. Operasi ini digunakan untuk menerapkan rumus ke data sumber dan menyimpan bidang hasil.

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

Masukkan pengaturan yang diperlukan.

Pengaturan	Deskripsi
Pilih rumus	Pilih rumus yang sudah ada dari menu dropdown atau pilih Kustom untuk memasukkan rumus secara manual.
Keluaran	Tentukan nama tampilan output untuk hasilnya.
Rumus	Masukkan rumus yang akan diterapkan ke data sumber.
Deskripsi	Berikan deskripsi untuk transformasi.
Nilai terakhir diketahui	Secara opsional, gunakan nilai terakhir yang diketahui jika nilai saat ini tidak tersedia.

Anda bisa memasukkan atau mengedit rumus di bidang Rumus . Rumus dapat menggunakan bidang dalam data sumber. Masukkan @ atau pilih Ctrl + Spasi untuk memilih titik data dari menu dropdown. Untuk rumus bawaan, pilih tempat penampung <dataflow> untuk melihat daftar titik data yang tersedia.

Anda dapat memasukkan properti metadata MQTT menggunakan format @$metadata.user_properties.<property> atau @$metadata.topic. Anda juga dapat memasukkan header $metadata menggunakan format @$metadata.<header>. Sintaks $metadata hanya diperlukan untuk properti MQTT yang merupakan bagian dari header pesan. Untuk informasi selengkapnya, lihat referensi bidang.

Rumus dapat menggunakan bidang dalam data sumber. Misalnya, Anda dapat menggunakan temperature bidang dalam data sumber untuk mengonversi suhu ke Celcius dan menyimpannya di temperatureCelsius bidang output.

Pilih Terapkan.

Ubah nama

Anda dapat mengganti nama titik data menggunakan transformasi Ganti Nama . Operasi ini digunakan untuk mengganti nama titik data dalam data sumber menjadi nama baru. Nama baru dapat digunakan dalam tahap berikutnya dari aliran data.

Di bawah Transformasi (opsional), pilih Ganti Nama>Tambahkan.

Masukkan pengaturan yang diperlukan.

Pengaturan	Deskripsi
Titik data	Pilih titik data dari menu dropdown atau masukkan header $metadata.
Nama titik data baru	Masukkan nama baru untuk titik data.
Deskripsi	Berikan deskripsi untuk transformasi.

Pilih Terapkan.

Properti baru

Anda dapat menambahkan properti baru ke data sumber menggunakan transformasi properti Baru. Operasi ini digunakan untuk menambahkan properti baru ke data sumber. Properti baru ini dapat digunakan pada tahap-tahap selanjutnya dari aliran data.

Di bawah Transformasi (opsional), pilih Tambahkan properti>baru.
Masukkan pengaturan yang diperlukan.

Pengaturan Deskripsi

Kunci properti Masukkan kunci untuk properti baru.

Nilai properti Masukkan nilai untuk properti baru.

Deskripsi Berikan deskripsi untuk properti baru.
Pilih Terapkan.

Pengaturan	Deskripsi
Kunci properti	Masukkan kunci untuk properti baru.
Nilai properti	Masukkan nilai untuk properti baru.
Deskripsi	Berikan deskripsi untuk properti baru.

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:

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

Anda dapat mengakses properti metadata MQTT menggunakan format $metadata.user_properties.<property> atau $metadata.topic. Anda juga dapat memasukkan header $metadata menggunakan format $metadata.<header>. Untuk informasi selengkapnya, lihat referensi bidang.

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

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.

Hapus

Secara default, semua titik data disertakan dalam skema output. Anda dapat menghapus titik data apa pun dari tujuan menggunakan transformasi Hapus .

Di bawah Transformasi (opsional), pilih Hapus.
Pilih titik data yang akan dihapus dari skema output.
Pilih Terapkan.

Untuk menghapus titik data dari skema output, Anda dapat menggunakan builtInTransformationSettings properti dalam konfigurasi aliran data. Properti map digunakan untuk menentukan titik data yang akan dihapus.

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

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

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

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.

Tips

Untuk menghasilkan skema dari file data sampel, gunakan Schema Gen Helper.

Untuk pengalaman operasi, Anda menentukan format skema dan serialisasi dalam detail titik akhir aliran data. Titik akhir yang mendukung format serialisasi adalah Microsoft Fabric OneLake, Azure Data Lake Storage Gen 2, Azure Data Explorer, dan penyimpanan lokal. Misalnya, untuk membuat serialisasi data dalam format Delta, Anda perlu mengunggah skema ke registri skema dan mereferensikannya dalam konfigurasi titik akhir tujuan aliran data.

Cuplikan layar menggunakan pengalaman operasi untuk mengatur serialisasi titik akhir tujuan aliran data.

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

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

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

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

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 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 tentang menggunakan titik akhir broker MQTT lokal, lihat Aliran data harus menggunakan titik akhir broker MQTT lokal.

Penting

Titik akhir penyimpanan memerlukan skema untuk serialisasi. Untuk menggunakan aliran data dengan Microsoft Fabric OneLake, Azure Data Lake Storage, Azure Data Explorer, atau Penyimpanan Lokal, Anda harus menentukan referensi skema.

Pilih titik akhir aliran data untuk digunakan sebagai tujuan.

Titik akhir penyimpanan memerlukan skema untuk serialisasi. Jika Anda memilih titik akhir tujuan Microsoft Fabric OneLake, Azure Data Lake Storage, Azure Data Explorer, atau Penyimpanan Lokal, Anda harus menentukan referensi skema. Misalnya, untuk menserialisasikan data ke titik akhir Microsoft Fabric dalam format Delta, Anda perlu mengunggah skema ke registri skema dan mereferensikannya dalam konfigurasi titik akhir tujuan aliran data.
Pilih Lanjutkan untuk mengonfigurasi tujuan.
Masukkan pengaturan yang diperlukan untuk tujuan, termasuk topik atau tabel untuk mengirim data. Lihat Mengonfigurasi tujuan data (topik, kontainer, atau tabel) untuk informasi selengkapnya.

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

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

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

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 Event Hubs, tujuan data adalah hub peristiwa individual dalam namespace tersebut.
Azure Data Lake Storage	Kontainer	Kontainer di akun penyimpanan. Bukan meja.
Microsoft Fabric OneLake	Tabel atau Folder	Sesuai dengan jenis jalur yang dikonfigurasi untuk endpoint.
Azure Data Explorer (layanan pencarian data)	Tabel	Tabel dalam database Azure Data Explorer.
Penyimpanan Lokal	Map	Nama folder atau direktori dalam pemasangan volume penyimpanan lokal yang persisten. Saat menggunakan Azure Container Storage yang diaktifkan oleh Azure Arc Cloud Ingest Edge Volumes, ini harus sesuai dengan parameter `spec.path` untuk subvolume yang Anda buat.

Untuk mengonfigurasi tujuan data:

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.

Cuplikan layar menampilkan pengalaman operasi yang meminta pengguna untuk memasukkan topik MQTT dengan jenis titik akhir tertentu.

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

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

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

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 menjadi Fahrenheit dan memfilter data di mana suhu dikalikan dengan kelembaban kurang dari 100000. Tujuan mengirimkan data ke topik factoryMQTT .

Gunakan perintah az iot ops dataflow apply untuk membuat atau mengubah aliran data.

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

Parameter --config-file adalah jalur dan nama file file konfigurasi JSON yang berisi properti sumber daya.

Dalam contoh ini, asumsikan file konfigurasi bernama data-flow.json dengan konten berikut yang disimpan di direktori beranda pengguna:

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

Berikut adalah contoh perintah untuk membuat atau memperbarui aliran data menggunakan profil aliran data default:

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

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

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

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

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

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

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

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

Untuk melihat lebih banyak contoh konfigurasi aliran data, lihat Azure REST API - Aliran data dan panduan 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.

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

Cuplikan layar antarmuka pengalaman operasi memperlihatkan opsi ekspor untuk aliran data yang dikonfigurasi.

Gunakan perintah az iot ops dataflow show untuk mengekspor aliran data.

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

Berikut adalah contoh perintah untuk mengekspor aliran data bernama data-flow ke file JSON bernama data-flow.json:

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

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 dataflow diatur ke 1 karena broker MQTT Event Grid tidak mendukung langganan bersama.
Saat menggunakan Azure Event Hubs sebagai sumber, setiap hub peristiwa di namespace adalah topik Kafka terpisah dan harus ditentukan sebagai sumber data.
Transformasi, jika digunakan, dikonfigurasi dengan sintaks yang tepat, termasuk penyertaan karakter khusus yang tepat.
Saat menggunakan titik akhir jenis penyimpanan sebagai tujuan, skema ditentukan.

Bagikan melalui

Membuat aliran data di Operasi Azure IoT

Prasyarat

Profil aliran data

Titik akhir aliran data

Mulai sekarang

Sumber

Opsi 1: Gunakan titik akhir broker pesan default sebagai sumber

Opsi 2: Gunakan aset sebagai sumber

Opsi 3: Gunakan titik akhir aliran data MQTT atau Kafka kustom sebagai sumber

Mengonfigurasi sumber data (topik MQTT atau Kafka)

Topik MQTT

Langganan bersama

Topik Kafka

Tentukan skema sumber

Transformasi

Pengayaan: Menambahkan data referensi

Filter: Memfilter data berdasarkan kondisi

Peta: Memindahkan data dari satu bidang ke bidang lainnya

Komputasi

Ubah nama

Properti baru

Hapus

Menserialisasikan data sesuai dengan skema

Tujuan

Mengonfigurasi tujuan data (topik, kontainer, atau tabel)

Contoh

Memverifikasi aliran data berfungsi

Mengekspor konfigurasi aliran data

Konfigurasi aliran data yang tepat

Langkah berikutnya

Saran dan Komentar

Sumber Daya Tambahan: