تكوين تدفقات البيانات في عمليات Azure IoT

هام

معاينة عمليات Azure IoT - التي تم تمكينها بواسطة Azure Arc قيد المعاينة حاليا. يجب عدم استخدام برنامج المعاينة هذا في بيئات الإنتاج.

ستحتاج إلى نشر تثبيت Azure IoT Operations جديد عند توفير إصدار متوفر بشكل عام. لن تتمكن من ترقية تثبيت معاينة.

للحصول على الشروط القانونية التي تنطبق على ميزات Azure الموجودة في الإصدار التجريبي، أو المعاينة، أو التي لم يتم إصدارها بعد في التوفر العام، راجع شروط الاستخدام التكميلية لمعاينات Microsoft Azure.

تدفق البيانات هو المسار الذي تأخذه البيانات من المصدر إلى الوجهة مع تحويلات اختيارية. يمكنك تكوين تدفق البيانات عن طريق إنشاء مورد مخصص ل Dataflow أو باستخدام مدخل Azure IoT Operations Studio. يتكون تدفق البيانات من ثلاثة أجزاء: المصدر والتحويل والوجهة.

لتعريف المصدر والوجهة، تحتاج إلى تكوين نقاط نهاية تدفق البيانات. التحويل اختياري ويمكن أن يتضمن عمليات مثل إثراء البيانات وتصفية البيانات وتعيين البيانات إلى حقل آخر.

هام

يجب أن يحتوي كل تدفق بيانات على نقطة النهاية الافتراضية لوسيط MQTT المحلي لعمليات Azure IoT كمصدر أو وجهة.

يمكنك استخدام تجربة العمليات في عمليات Azure IoT لإنشاء تدفق بيانات. توفر تجربة العمليات واجهة مرئية لتكوين تدفق البيانات. يمكنك أيضا استخدام Bicep لإنشاء تدفق بيانات باستخدام ملف قالب Bicep، أو استخدام Kubernetes لإنشاء تدفق بيانات باستخدام ملف YAML.

تابع القراءة لمعرفة كيفية تكوين المصدر والتحويل والوجهة.

المتطلبات الأساسية

يمكنك نشر تدفقات البيانات بمجرد أن يكون لديك مثيل Azure IoT Operations Preview باستخدام ملف تعريف تدفق البيانات الافتراضي ونقطة النهاية. ومع ذلك، قد تحتاج إلى تكوين ملفات تعريف تدفق البيانات ونقاط النهاية لتخصيص تدفق البيانات.

ملف تعريف تدفق البيانات

يحدد ملف تعريف تدفق البيانات عدد المثيلات لتدفقات البيانات ضمنها لاستخدامها. إذا لم تكن بحاجة إلى مجموعات متعددة من تدفقات البيانات بإعدادات تحجيم مختلفة، يمكنك استخدام ملف تعريف تدفق البيانات الافتراضي. لمعرفة كيفية تكوين ملف تعريف تدفق البيانات، راجع تكوين ملفات تعريف تدفق البيانات.

نقاط نهاية تدفق البيانات

نقاط نهاية تدفق البيانات مطلوبة لتكوين المصدر والوجهة لتدفق البيانات. للبدء بسرعة، يمكنك استخدام نقطة نهاية تدفق البيانات الافتراضية لوسيط MQTT المحلي. يمكنك أيضا إنشاء أنواع أخرى من نقاط نهاية تدفق البيانات مثل Kafka أو Event Hubs أو Azure Data Lake Storage. لمعرفة كيفية تكوين كل نوع من نقاط نهاية تدفق البيانات، راجع تكوين نقاط نهاية تدفق البيانات.

الشروع في العمل

بمجرد أن يكون لديك المتطلبات الأساسية، يمكنك البدء في إنشاء تدفق بيانات.

بوابة

لإنشاء تدفق بيانات في تجربة العمليات، حدد Dataflow>Create dataflow. ثم ترى الصفحة حيث يمكنك تكوين المصدر والتحويل والوجهة لتدفق البيانات.

Bicep

إنشاء ملف Bicep .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
}

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

إنشاء ملف بيان .yaml Kubernetes لبدء إنشاء تدفق بيانات. يوضح هذا المثال بنية تدفق البيانات التي تحتوي على تكوينات المصدر والتحويل والوجهة.

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

راجع الأقسام التالية لمعرفة كيفية تكوين أنواع عمليات تدفق البيانات.

المصدر

لتكوين مصدر لتدفق البيانات، حدد مرجع نقطة النهاية وقائمة بمصادر البيانات لنقطة النهاية.

استخدام الأصل كمصدر

بوابة

يمكنك استخدام أصل كمصدر لتدفق البيانات. لا يتوفر استخدام أحد الأصول كمصدر إلا في تجربة العمليات.

1. ضمن Source details، حدد Asset.

2. حدد الأصل الذي تريد استخدامه كنقطة نهاية المصدر.

3. حدد Proceed.

   يتم عرض قائمة بنقاط البيانات للأصل المحدد.

   

4. حدد Apply لاستخدام الأصل كنقطة نهاية المصدر.

Bicep

تكوين أصل كمصدر متاح فقط في تجربة العمليات.

Kubernetes

تكوين أصل كمصدر متاح فقط في تجربة العمليات.

عند استخدام أصل كمصدر، يتم استخدام تعريف الأصل للاستدلال على مخطط تدفق البيانات. يتضمن تعريف الأصل مخطط نقاط بيانات الأصل. لمعرفة المزيد، راجع إدارة تكوينات الأصول عن بعد.

بمجرد تكوينها، وصلت البيانات من الأصل إلى تدفق البيانات عبر وسيط MQTT المحلي. لذلك، عند استخدام أصل كمصدر، يستخدم تدفق البيانات نقطة النهاية الافتراضية ل وسيط MQTT المحلي كمصدر في الواقع.

استخدام نقطة نهاية MQTT الافتراضية كمصدر

بوابة

1. ضمن Source details، حدد MQTT.

   

2. أدخل الإعدادات التالية لمصدر MQTT:

   الإعدادات	‏‏الوصف
   موضوع MQTT	عامل تصفية موضوع MQTT للاشتراك فيه للرسائل الواردة. راجع تكوين مواضيع MQTT أو Kafka.
   مخطط الرسائل	المخطط الذي يجب استخدامه لإلغاء تسلسل الرسائل الواردة. راجع تحديد مخطط لإلغاء تسلسل البيانات.

3. حدد تطبيق.

Bicep

يتم تكوين نقطة نهاية MQTT في ملف قالب Bicep. على سبيل المثال، نقطة النهاية التالية هي مصدر لتدفق البيانات.

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

هنا، dataSources اسمح لك بتحديد مواضيع MQTT أو Kafka متعددة دون الحاجة إلى تعديل تكوين نقطة النهاية. تعني هذه المرونة أنه يمكن إعادة استخدام نفس نقطة النهاية عبر تدفقات بيانات متعددة، حتى إذا كانت الموضوعات تختلف. لمعرفة المزيد، راجع تكوين مصادر البيانات.

Kubernetes

على سبيل المثال، لتكوين مصدر باستخدام نقطة نهاية MQTT واثنين من عوامل تصفية موضوع MQTT، استخدم التكوين التالي:

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

لأنه dataSources يسمح لك بتحديد مواضيع MQTT أو Kafka دون تعديل تكوين نقطة النهاية، يمكنك إعادة استخدام نقطة النهاية لتدفقات بيانات متعددة حتى إذا كانت الموضوعات مختلفة. لمعرفة المزيد، راجع تكوين مصادر البيانات.

إذا لم يتم استخدام نقطة النهاية الافتراضية كمصدر، فيجب استخدامها كوجهة. لمعرفة المزيد حول، راجع يجب أن تستخدم تدفقات البيانات نقطة نهاية وسيط MQTT المحلية.

استخدام نقطة نهاية تدفق بيانات MQTT أو Kafka المخصصة كمصدر

إذا قمت بإنشاء نقطة نهاية تدفق بيانات MQTT أو Kafka مخصصة (على سبيل المثال، لاستخدامها مع Event Grid أو Event Hubs)، يمكنك استخدامها كمصدر لتدفق البيانات. تذكر أنه لا يمكن استخدام نقاط نهاية نوع التخزين، مثل Data Lake أو Fabric OneLake، كمصدر.

لتكوين، استخدم Kubernetes YAML أو Bicep. استبدل قيم العنصر النائب باسم نقطة النهاية المخصصة والموضوعات.

بوابة

استخدام نقطة نهاية MQTT أو Kafka مخصصة كمصدر غير مدعوم حاليا في تجربة العمليات.

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

تكوين مصادر البيانات (موضوعات MQTT أو Kafka)

يمكنك تحديد مواضيع MQTT أو Kafka متعددة في مصدر دون الحاجة إلى تعديل تكوين نقطة نهاية تدفق البيانات. تعني هذه المرونة أنه يمكن إعادة استخدام نفس نقطة النهاية عبر تدفقات بيانات متعددة، حتى إذا كانت الموضوعات تختلف. لمزيد من المعلومات، راجع إعادة استخدام نقاط نهاية تدفق البيانات.

موضوعات MQTT

عندما يكون المصدر نقطة نهاية MQTT (مضمنة في شبكة الأحداث)، يمكنك استخدام عامل تصفية موضوع MQTT للاشتراك في الرسائل الواردة. يمكن أن يتضمن عامل تصفية الموضوع أحرف بدل للاشتراك في مواضيع متعددة. على سبيل المثال، thermostats/+/telemetry/temperature/# يشترك في جميع رسائل القياس عن بعد لدرجة الحرارة من منظمات الحرارة. لتكوين عوامل تصفية موضوع MQTT:

بوابة

في تفاصيل مصدر تدفق البيانات لتجربة العمليات، حدد MQTT، ثم استخدم حقل موضوع MQTT لتحديد عامل تصفية موضوع MQTT للاشتراك فيه للرسائل الواردة.

إشعار

يمكن تحديد عامل تصفية موضوع MQTT واحد فقط في تجربة العمليات. لاستخدام عوامل تصفية مواضيع MQTT متعددة، استخدم Bicep أو Kubernetes.

Bicep

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

مثال مع عوامل تصفية موضوع MQTT متعددة مع أحرف البدل:

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

هنا، يتم استخدام حرف + البدل لتحديد جميع الأجهزة ضمن thermostats المواضيع و humidifiers . # يتم استخدام حرف البدل لتحديد جميع رسائل بيانات تتبع الاستخدام ضمن جميع المواضيع الفرعية للمواضيع temperature وhumidity.

Kubernetes

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

مثال مع عوامل تصفية موضوع MQTT متعددة مع أحرف البدل:

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

هنا، يتم استخدام حرف + البدل لتحديد جميع الأجهزة ضمن thermostats المواضيع و humidifiers . # يتم استخدام حرف البدل لتحديد جميع رسائل بيانات تتبع الاستخدام ضمن جميع المواضيع الفرعية للمواضيع temperature وhumidity.

الاشتراكات المشتركة

لاستخدام الاشتراكات المشتركة مع مصادر MQTT، يمكنك تحديد موضوع الاشتراك المشترك في شكل $shared/<GROUP_NAME>/<TOPIC_FILTER>.

بوابة

في تفاصيل مصدر تدفق البيانات لتجربة العمليات، حدد MQTT واستخدم حقل موضوع MQTT لتحديد مجموعة الاشتراك المشترك والموضوع.

Bicep

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

Kubernetes

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

إذا كان عدد المثيلات في ملف تعريف تدفق البيانات أكبر من 1، يتم تمكين الاشتراك المشترك تلقائيا لجميع تدفقات البيانات التي تستخدم مصدر MQTT. في هذه الحالة، تتم إضافة البادئة $shared ويتم إنشاء اسم مجموعة الاشتراك المشترك تلقائيا. على سبيل المثال، إذا كان لديك ملف تعريف تدفق بيانات مع عدد مثيلات 3، ويستخدم تدفق البيانات نقطة نهاية MQTT كمصدر تم تكوينه مع الموضوعات topic1 و topic2، يتم تحويلها تلقائيا إلى اشتراكات مشتركة ك $shared/<GENERATED_GROUP_NAME>/topic1 و $shared/<GENERATED_GROUP_NAME>/topic2. إذا كنت تريد استخدام معرف مجموعة اشتراك مشترك مختلف، يمكنك تجاوزه في الموضوع، مثل $shared/mygroup/topic1.

هام

تعد تدفقات البيانات التي تتطلب اشتراكا مشتركا عندما يكون عدد المثيلات أكبر من 1 أمرا مهما عند استخدام وسيط MQTT لشبكة الأحداث كمصدر لأنه لا يدعم الاشتراكات المشتركة. لتجنب فقدان الرسائل، قم بتعيين عدد مثيلات ملف تعريف تدفق البيانات إلى 1 عند استخدام وسيط MQTT لشبكة الأحداث كمصدر. وذلك عندما يكون تدفق البيانات هو المشترك ويتلقى الرسائل من السحابة.

موضوعات Kafka

عندما يكون المصدر نقطة نهاية Kafka (مضمنة في مراكز الأحداث)، حدد مواضيع kafka الفردية للاشتراك فيها للرسائل الواردة. أحرف البدل غير معتمدة، لذلك يجب تحديد كل موضوع بشكل ثابت.

إشعار

عند استخدام مراكز الأحداث عبر نقطة نهاية Kafka، يكون كل مركز أحداث فردي داخل مساحة الاسم هو موضوع Kafka. على سبيل المثال، إذا كان لديك مساحة اسم لمراكز الأحداث مع مركزي أحداث، وhumidifiers، thermostats يمكنك تحديد كل مركز أحداث كموضوع Kafka.

لتكوين مواضيع Kafka:

بوابة

استخدام نقطة نهاية Kafka كمصدر غير مدعوم حاليا في تجربة العمليات.

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

تحديد مخطط لإلغاء تسلسل البيانات

إذا كانت البيانات المصدر تحتوي على حقول أو حقول اختيارية ذات أنواع مختلفة، فحدد مخطط إلغاء التسلسل لضمان التناسق. على سبيل المثال، قد تحتوي البيانات على حقول غير موجودة في كافة الرسائل. بدون المخطط، لن يتمكن التحويل من معالجة هذه الحقول حيث سيكون لها قيم فارغة. باستخدام المخطط، يمكنك تحديد القيم الافتراضية أو تجاهل الحقول.

يعد تحديد المخطط مناسبا فقط عند استخدام مصدر MQTT أو Kafka. إذا كان المصدر أحد الأصول، يتم استنتاج المخطط تلقائيا من تعريف الأصل.

لتكوين المخطط المستخدم لإلغاء تسلسل الرسائل الواردة من مصدر:

بوابة

في العمليات تجربة تفاصيل مصدر تدفق البيانات، حدد MQTT واستخدم حقل مخطط الرسالة لتحديد المخطط. يمكنك استخدام الزر Upload لتحميل ملف مخطط أولا. لمعرفة المزيد، راجع فهم مخططات الرسائل.

Bicep

بمجرد استخدام سجل المخطط لتخزين المخطط، يمكنك الرجوع إليه في تكوين تدفق البيانات.

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

Kubernetes

بمجرد استخدام سجل المخطط لتخزين المخطط، يمكنك الرجوع إليه في تكوين تدفق البيانات.

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

التحويل

عملية التحويل هي المكان الذي يمكنك فيه تحويل البيانات من المصدر قبل إرسالها إلى الوجهة. التحويلات اختيارية. إذا لم تكن بحاجة إلى إجراء تغييرات على البيانات، فلا تقم بتضمين عملية التحويل في تكوين تدفق البيانات. يتم ربط تحويلات متعددة معا على مراحل بغض النظر عن الترتيب الذي يتم تحديدها به في التكوين. ترتيب المراحل هو دائما:

1. إثراء خاصية جديدة أو إعادة تسميتها أو إضافتها: أضف بيانات إضافية إلى البيانات المصدر نظرا لمجموعة بيانات وشرط لمطابقته.
2. عامل التصفية: تصفية البيانات استنادا إلى شرط.
3. تعيين أو حساب: نقل البيانات من حقل إلى آخر مع تحويل اختياري.

بوابة

في تجربة العمليات، حدد Dataflow>Add transform (اختياري) .

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

إثراء: إضافة بيانات مرجعية

لإثراء البيانات، يمكنك استخدام مجموعة البيانات المرجعية في مخزن الحالة الموزعة لعمليات Azure IoT (DSS). يتم استخدام مجموعة البيانات لإضافة بيانات إضافية إلى بيانات المصدر استنادا إلى شرط. يتم تحديد الشرط ك حقل في البيانات المصدر التي تطابق حقلا في مجموعة البيانات.

يمكنك تحميل نموذج البيانات في DSS باستخدام نموذج أداة مجموعة DSS. تتوافق الأسماء الرئيسية في مخزن الحالة الموزعة مع مجموعة بيانات في تكوين تدفق البيانات.

بوابة

في تجربة العمليات، يتم دعم مرحلة الإثراء حاليا باستخدام تحويلات خاصية Rename و New.

1. في تجربة العمليات، حدد تدفق بيانات ثم إضافة تحويل (اختياري) .

2. اختر Rename أو New property transforms ثم حدد Add.

   

Bicep

يوضح هذا المثال كيف يمكنك استخدام الحقل في deviceId البيانات المصدر لمطابقة asset الحقل في مجموعة البيانات:

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

Kubernetes

على سبيل المثال، يمكنك استخدام deviceId الحقل في البيانات المصدر لمطابقة asset الحقل في مجموعة البيانات:

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

إذا كانت مجموعة البيانات تحتوي على سجل مع asset الحقل، على غرار:

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

تحتوي البيانات من المصدر مع deviceId مطابقة thermostat1 الحقل على location الحقلين و manufacturer المتوفرين في مرحلتي التصفية والخريطة.

لمزيد من المعلومات حول بناء جملة الشرط، راجع إثراء البيانات باستخدام تدفقات البيانات وتحويل البيانات باستخدام تدفقات البيانات.

عامل التصفية: تصفية البيانات استنادا إلى شرط

لتصفية البيانات على شرط، يمكنك استخدام filter المرحلة. يتم تحديد الشرط ك حقل في البيانات المصدر التي تطابق قيمة.

بوابة

1. ضمن Transform (اختياري)، حدد Filter>Add.

2. اختر نقاط البيانات لتضمينها في مجموعة البيانات.

3. إضافة شرط عامل تصفية ووصف.

   

4. حدد تطبيق.

على سبيل المثال، يمكنك استخدام شرط عامل تصفية مثل temperature > 20 تصفية البيانات الأقل من أو تساوي 20 استنادا إلى حقل درجة الحرارة.

Bicep

على سبيل المثال، يمكنك استخدام temperature الحقل في البيانات المصدر لتصفية البيانات:

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

temperature إذا كان الحقل أكبر من 20، يتم تمرير البيانات إلى المرحلة التالية. temperature إذا كان الحقل أقل من أو يساوي 20، تتم تصفية البيانات.

Kubernetes

على سبيل المثال، يمكنك استخدام temperature الحقل في البيانات المصدر لتصفية البيانات:

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

temperature إذا كان الحقل أكبر من 20، يتم تمرير البيانات إلى المرحلة التالية. temperature إذا كان الحقل أقل من أو يساوي 20، تتم تصفية البيانات.

الخريطة: نقل البيانات من حقل إلى آخر

لتعيين البيانات إلى حقل آخر مع تحويل اختياري، يمكنك استخدام map العملية. يتم تحديد التحويل كصيغة تستخدم الحقول في البيانات المصدر.

بوابة

في تجربة العمليات، يتم دعم التعيين حاليا باستخدام تحويلات الحساب .

1. ضمن Transform (اختياري) ، حدد Compute>Add.

2. أدخل الحقول والتعبيرات المطلوبة.

   

3. حدد تطبيق.

Bicep

على سبيل المثال، يمكنك استخدام temperature الحقل في البيانات المصدر لتحويل درجة الحرارة إلى درجة مئوية وتخزينها في temperatureCelsius الحقل. يمكنك أيضا إثراء بيانات المصدر بالحقل location من مجموعة بيانات السياق:

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

Kubernetes

على سبيل المثال، يمكنك استخدام temperature الحقل في البيانات المصدر لتحويل درجة الحرارة إلى درجة مئوية وتخزينها في temperatureCelsius الحقل. يمكنك أيضا إثراء بيانات المصدر بالحقل location من مجموعة بيانات السياق:

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

لمعرفة المزيد، راجع تعيين البيانات باستخدام تدفقات البيانات وتحويل البيانات باستخدام تدفقات البيانات.

تسلسل البيانات وفقا لمخطط

إذا كنت تريد تسلسل البيانات قبل إرسالها إلى الوجهة، فستحتاج إلى تحديد تنسيق مخطط وتسلسل. وإلا، يتم تسلسل البيانات في JSON مع الأنواع المستنتجة. تتطلب نقاط نهاية التخزين مثل Microsoft Fabric أو Azure Data Lake مخططا لضمان تناسق البيانات. تنسيقات التسلسل المدعومة هي Parquet و Delta.

بوابة

حاليا، تحديد مخطط الإخراج والتسلسل غير مدعوم في تجربة العمليات.

Bicep

بمجرد تحميل مخطط إلى سجل المخطط، يمكنك الرجوع إليه في تكوين تدفق البيانات.

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

Kubernetes

بمجرد تحميل مخطط إلى سجل المخطط، يمكنك الرجوع إليه في تكوين تدفق البيانات.

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

لمزيد من المعلومات حول سجل المخطط، راجع فهم مخططات الرسائل.

الوجهة

لتكوين وجهة لتدفق البيانات، حدد مرجع نقطة النهاية ووجهة البيانات. يمكنك تحديد قائمة وجهات البيانات لنقطة النهاية.

لإرسال البيانات إلى وجهة أخرى غير وسيط MQTT المحلي، قم بإنشاء نقطة نهاية تدفق البيانات. لمعرفة كيفية إجراء ذلك، راجع تكوين نقاط نهاية تدفق البيانات. إذا لم تكن الوجهة وسيط MQTT المحلي، فيجب استخدامها كمصدر. لمعرفة المزيد حول، راجع يجب أن تستخدم تدفقات البيانات نقطة نهاية وسيط MQTT المحلية.

هام

تتطلب نقاط نهاية التخزين مرجع مخطط. إذا قمت بإنشاء نقاط نهاية وجهة التخزين ل Microsoft Fabric OneLake وADS Gen 2 وAzure Data Explorer والتخزين المحلي، يجب تحديد مرجع المخطط.

بوابة

1. حدد نقطة نهاية تدفق البيانات لاستخدامها كوجهة.

   

2. حدد متابعة لتكوين الوجهة.

3. أدخل الإعدادات المطلوبة للوجهة، بما في ذلك الموضوع أو الجدول لإرسال البيانات إليه. راجع تكوين وجهة البيانات (موضوع أو حاوية أو جدول) لمزيد من المعلومات.

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

تكوين وجهة البيانات (موضوع أو حاوية أو جدول)

على غرار مصادر البيانات، فإن وجهة البيانات هي مفهوم يستخدم للحفاظ على نقاط نهاية تدفق البيانات قابلة لإعادة الاستخدام عبر تدفقات بيانات متعددة. بشكل أساسي، يمثل الدليل الفرعي في تكوين نقطة نهاية تدفق البيانات. على سبيل المثال، إذا كانت نقطة نهاية تدفق البيانات هي نقطة نهاية تخزين، فإن وجهة البيانات هي الجدول في حساب التخزين. إذا كانت نقطة نهاية تدفق البيانات هي نقطة نهاية Kafka، فإن وجهة البيانات هي موضوع Kafka.

نوع نقطة النهاية	معنى وجهة البيانات	‏‏الوصف
MQTT (أو شبكة الأحداث)	الموضوع	موضوع MQTT حيث يتم إرسال البيانات. يتم دعم الموضوعات الثابتة فقط، ولا توجد أحرف بدل.
Kafka (أو مراكز الأحداث)	الموضوع	موضوع Kafka حيث يتم إرسال البيانات. يتم دعم الموضوعات الثابتة فقط، ولا توجد أحرف بدل. إذا كانت نقطة النهاية عبارة عن مساحة اسم لمراكز الأحداث، فإن وجهة البيانات هي مركز الأحداث الفردي داخل مساحة الاسم.
Azure Data Lake Storage Gen2	الحاوية	الحاوية في حساب التخزين. ليس الجدول.
Microsoft Fabric OneLake	جدول أو مجلد	يتوافق مع نوع المسار المكون لنقطة النهاية.
Azure Data Explorer ‏(Kusto)	جدول	الجدول في قاعدة بيانات Azure Data Explorer.
التخزين المحلي	مجلد	اسم المجلد أو الدليل في تحميل وحدة التخزين الثابتة للتخزين المحلي. عند استخدام Azure Container Storage الذي تم تمكينه بواسطة Azure Arc Cloud Ingest Edge Volumes، يجب أن يتطابق هذا مع spec.path المعلمة للوحدة الفرعية التي قمت بإنشائها.

لتكوين وجهة البيانات:

بوابة

عند استخدام تجربة العمليات، يتم تفسير حقل وجهة البيانات تلقائيا استنادا إلى نوع نقطة النهاية. على سبيل المثال، إذا كانت نقطة نهاية تدفق البيانات نقطة نهاية تخزين، فستطالبك صفحة تفاصيل الوجهة بإدخال اسم الحاوية. إذا كانت نقطة نهاية تدفق البيانات هي نقطة نهاية MQTT، فستطالبك صفحة تفاصيل الوجهة بإدخال الموضوع، وهكذا.

Bicep

بناء الجملة هو نفسه لجميع نقاط نهاية تدفق البيانات:

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

على سبيل المثال، لإرسال البيانات مرة أخرى إلى وسيط MQTT المحلي موضوع MQTT ثابت، استخدم التكوين التالي:

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

أو، إذا كان لديك نقطة نهاية مخصصة لمركز الأحداث، فسيبدو التكوين كما يلي:

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

مثال آخر باستخدام نقطة نهاية التخزين كوجهة:

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

Kubernetes

بناء الجملة هو نفسه لجميع نقاط نهاية تدفق البيانات:

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

على سبيل المثال، لإرسال البيانات مرة أخرى إلى وسيط MQTT المحلي موضوع MQTT ثابت، استخدم التكوين التالي:

destinationSettings:
  endpointRef: default
  dataDestination: example-topic

أو، إذا كان لديك نقطة نهاية مخصصة لمركز الأحداث، فسيبدو التكوين كما يلي:

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

مثال آخر باستخدام نقطة نهاية التخزين كوجهة:

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

مثال

المثال التالي هو تكوين تدفق البيانات الذي يستخدم نقطة نهاية MQTT للمصدر والوجهة. يقوم المصدر بتصفية البيانات من موضوع azure-iot-operations/data/thermostatMQTT . يحول التحويل درجة الحرارة إلى فهرنهايت ويصفي البيانات حيث تكون درجة الحرارة مضروبة في الرطوبة أقل من 100000. ترسل الوجهة البيانات إلى موضوع factoryMQTT .

بوابة

راجع علامات تبويب Bicep أو Kubernetes للحصول على مثال التكوين.

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

لمشاهدة المزيد من الأمثلة على تكوينات تدفق البيانات، راجع Azure REST API - Dataflow وBicep التشغيل السريع.

التحقق من عمل تدفق البيانات

اتبع البرنامج التعليمي: جسر MQTT ثنائي الاتجاه إلى Azure Event Grid للتحقق من عمل تدفق البيانات.

تصدير تكوين تدفق البيانات

لتصدير تكوين تدفق البيانات، يمكنك استخدام تجربة العمليات أو عن طريق تصدير المورد المخصص ل Dataflow.

بوابة

حدد تدفق البيانات الذي تريد تصديره وحدد تصدير من شريط الأدوات.

Bicep

Bicep هي بنية أساسية كتعلم برمجي ولا يلزم تصدير. استخدم ملف قالب Bicep لإنشاء تدفق بيانات لإعداد تدفقات البيانات وتكوينها بسرعة.

Kubernetes

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

تكوين تدفق البيانات المناسب

للتأكد من أن تدفق البيانات يعمل كما هو متوقع، تحقق مما يلي:

• يجب استخدام نقطة نهاية تدفق بيانات MQTT الافتراضية إما كمصدر أو وجهة.
• ملف تعريف تدفق البيانات موجود ومشار إليه في تكوين تدفق البيانات.
• المصدر هو إما نقطة نهاية MQTT أو نقطة نهاية Kafka أو أصل. لا يمكن استخدام نقاط نهاية نوع التخزين كمصدر.
• عند استخدام Event Grid كمصدر، يتم تعيين عدد مثيلات ملف تعريف تدفق البيانات إلى 1 لأن وسيط Event Grid MQTT لا يدعم الاشتراكات المشتركة.
• عند استخدام Event Hubs كمصدر، يكون كل مركز أحداث في مساحة الاسم موضوع Kafka منفصل ويجب تحديده كمصدر بيانات.
• يتم تكوين التحويل، إذا تم استخدامه، مع بناء جملة مناسب، بما في ذلك الهروب المناسب من الأحرف الخاصة.
• عند استخدام نقاط نهاية نوع التخزين كوجهة ، يتم تحديد مخطط.

الخطوات التالية

• تعيين البيانات باستخدام تدفقات البيانات
• تحويل البيانات باستخدام تدفقات البيانات
• إثراء البيانات باستخدام تدفقات البيانات
• فهم مخططات الرسائل
• إدارة ملفات تعريف تدفق البيانات