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.

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

Artikel ini memperlihatkan kepada Anda cara membuat aliran data dengan contoh, termasuk sumber, transformasi, dan tujuan.

Prasyarat

• Instans Pratinjau Operasi Azure IoT
• Profil aliran data yang dikonfigurasi
• Titik akhir aliran data. Misalnya, buat titik akhir aliran data untuk broker MQTT lokal. Anda dapat menggunakan titik akhir ini untuk sumber dan tujuan. Atau, Anda dapat mencoba titik akhir lain 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.

Buat aliran data

Setelah Anda memiliki titik akhir aliran data, Anda bisa menggunakannya untuk membuat aliran data. Ingat bahwa aliran data terdiri dari tiga bagian: sumber, transformasi, dan tujuan.

Portal

Untuk membuat aliran data di portal pengalaman operasi, pilih Aliran Data>Buat aliran data.

Kubernetes

Struktur keseluruhan konfigurasi aliran data adalah sebagai berikut:

apiVersion: connectivity.iotoperations.azure.com/v1beta1
kind: Dataflow
metadata:
  name: my-dataflow
  namespace: azure-iot-operations
spec:
  profileRef: default
  mode: Enabled
  operations:
    - operationType: Source
      sourceSettings:
        # See source configuration section
    - 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.

Mengonfigurasi sumber dengan titik akhir aliran data untuk mendapatkan data

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

Portal

Menggunakan Aset sebagai sumber

Anda dapat menggunakan aset sebagai sumber untuk aliran data. Ini hanya tersedia di portal 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.

Kubernetes

Mengonfigurasi aset sebagai sumber hanya tersedia di portal pengalaman operasi.

Menggunakan MQTT sebagai sumber

Portal

1. Di bawah Detail sumber, pilih MQTT.

2. Masukkan Topik MQTT yang ingin Anda dengarkan untuk pesan masuk.

3. Pilih Skema Pesan dari daftar dropdown atau unggah skema baru. 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.

   

4. Pilih Terapkan.

Kubernetes

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

sourceSettings:
  endpointRef: mq
  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 Menggunakan kembali titik akhir aliran data.

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.

spec:
  operations:
  - operationType: Source
    sourceSettings:
      serializationFormat: Json
      schemaRef: aio-sr://exampleNamespace/exampleAvroSchema:1.0.0

Untuk menentukan skema, buat file dan simpan di registri skema.

{
  "type": "record",
  "name": "Temperature",
  "fields": [
    {"name": "deviceId", "type": "string"},
    {"name": "temperature", "type": "float"}
  ]
}

Catatan

Satu-satunya format serialisasi yang didukung adalah JSON. Skema bersifat opsional.

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

Langganan bersama

Untuk menggunakan langganan bersama dengan sumber MQTT, Anda dapat menentukan topik langganan bersama dalam bentuk $shared/<subscription-group>/<topic>.

sourceSettings:
  dataSources:
    - $shared/myGroup/thermostats/+/telemetry/temperature/#

Catatan

Jika jumlah instans di profil aliran data lebih besar dari 1, maka topik langganan bersama harus digunakan.

Mengonfigurasi transformasi untuk memproses data

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. Pengayaan: Tambahkan data tambahan ke data sumber yang diberi himpunan data dan kondisi yang cocok.
2. Filter: Memfilter data berdasarkan kondisi.
3. Peta: Memindahkan data dari satu bidang ke bidang lain dengan konversi opsional.

Portal

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

Kubernetes

builtInTransformationSettings:
  datasets:
    # ...
  filter:
    # ...
  map:
    # ...

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.

Nama kunci di penyimpanan status terdistribusi sesuai dengan himpunan data dalam konfigurasi aliran data.

Portal

Saat ini, operasi pengayaan tidak tersedia di portal pengalaman operasi.

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 filter tahap dan map .

Anda dapat memuat data sampel ke DSS dengan menggunakan sampel alat set DSS.

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.

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

Di portal 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.

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
      output: temperatureCelsius
      expression: "($1 - 32) * 5/9"
    - 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. Ingatlah bahwa titik akhir penyimpanan seperti Microsoft Fabric atau Azure Data Lake memerlukan skema untuk memastikan konsistensi data.

Portal

Tentukan skema Output saat Anda menambahkan titik akhir aliran data tujuan.

Kubernetes

builtInTransformationSettings:
  serializationFormat: Parquet
  schemaRef: aio-sr://<NAMESPACE>/<SCHEMA>:<VERSION>

Untuk menentukan skema, Anda dapat membuat sumber daya kustom Skema dengan definisi skema.

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

{
  "type": "record",
  "name": "Temperature",
  "fields": [
    {"name": "deviceId", "type": "string"},
    {"name": "temperatureCelsius", "type": "float"}
    {"name": "location", "type": "string"}
  ]
}

Format serialisasi yang didukung adalah JSON, Parquet, dan Delta.

Mengonfigurasi tujuan dengan titik akhir aliran data untuk mengirim data

Untuk mengonfigurasi tujuan untuk aliran data, tentukan referensi titik akhir dan tujuan data. Anda dapat menentukan daftar tujuan data untuk titik akhir yang merupakan topik MQTT atau Kafka.

Portal

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

   

2. Pilih Lanjutkan untuk mengonfigurasi tujuan.

3. Tambahkan detail pemetaan berdasarkan jenis tujuan.

Kubernetes

Misalnya, untuk mengonfigurasi tujuan menggunakan titik akhir MQTT yang dibuat sebelumnya dan topik MQTT statis, gunakan konfigurasi berikut:

destinationSettings:
  endpointRef: mq
  dataDestination: factory

Jika Anda telah membuat titik akhir penyimpanan seperti Microsoft Fabric, gunakan bidang tujuan data untuk menentukan nama tabel atau kontainer:

destinationSettings:
  endpointRef: adls
  dataDestination: telemetryTable

Contoh

Contoh berikut adalah konfigurasi aliran data yang menggunakan titik akhir MQTT untuk sumber dan tujuan. Sumber memfilter data dari topik thermostats/+/telemetry/temperature/# MQTT dan humidifiers/+/telemetry/humidity/#. Transformasi mengonversi suhu ke Fahrenheit dan memfilter data di mana suhu kurang dari 100000. Tujuan mengirimkan data ke topik factoryMQTT .

apiVersion: connectivity.iotoperations.azure.com/v1beta1
kind: Dataflow
metadata:
  name: my-dataflow
  namespace: azure-iot-operations
spec:
  profileRef: default
  mode: Enabled
  operations:
    - operationType: Source
      sourceSettings:
        endpointRef: mq
        dataSources:
          - thermostats/+/telemetry/temperature/#
          - humidifiers/+/telemetry/humidity/#
    - operationType: builtInTransformation
      builtInTransformationSettings:
        filter:
          - inputs:
              - 'temperature.Value'
              - '"Tag 10".Value'
            expression: "$1*$2<100000"
        map:
          - inputs:
              - '*'
            output: '*'
          - inputs:
              - temperature.Value
            output: TemperatureF
            expression: cToF($1)
          - inputs:
              - '"Tag 10".Value'
            output: 'Tag 10'
    - operationType: Destination
      destinationSettings:
        endpointRef: mq
        dataDestination: factory

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 portal pengalaman operasi atau dengan mengekspor sumber daya kustom Aliran Data.

Portal

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

Kubernetes

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