Настройте конечные точки потока данных для Azure Event Hubs и Kafka

Чтобы настроить двунаправленное взаимодействие между операциями Интернета вещей Azure и брокерами Apache Kafka, можно настроить конечную точку потока данных. Эта конфигурация позволяет указать конечную точку соединения, TLS, проверку подлинности и другие параметры.

Предварительные условия

• Экземпляр операций Интернета вещей Azure

Центры событий Azure

Центры событий Azure совместим с протоколом Kafka и может использоваться с потоками данных с некоторыми ограничениями.

Создание пространства имен Центры событий Azure и концентратора событий

Сначала создайте пространство имен с поддержкой Kafka Центры событий Azure

Затем создайте концентратор событий в пространстве имен. Каждый отдельный концентратор событий соответствует теме Kafka. Вы можете создать несколько центров событий в одном пространстве имен, чтобы представить несколько разделов Kafka.

Назначьте разрешение управляемому удостоверению

Чтобы настроить конечную точку потока данных для Центра событий Azure, мы рекомендуем использовать управляемое удостоверение, назначенное пользователем или системой. Этот подход является безопасным и устраняет необходимость управления учетными данными вручную.

После создания пространства имен и концентратора событий в Azure Event Hubs необходимо назначить роль управляемому удостоверению Azure IoT Operations, которая предоставляет разрешение на отправку и получение сообщений в концентратор событий.

При использовании управляемого удостоверения, назначаемого системой, в портал Azure перейдите к экземпляру Операций Интернета вещей Azure и выберите "Обзор". Скопируйте имя расширения, указанного после расширения Azure IoT Operations Arc. Например, azure-iot-operations-xxxx7. Управляемое удостоверение, назначаемое системой, можно найти с помощью того же имени расширения Azure Arc для операций IoT.

Затем перейдите к пространству имен Центров событий >управление доступом (IAM)>добавить назначение ролей.

1. На вкладке "Роль" выберите соответствующую роль, например Azure Event Hubs Data Sender или Azure Event Hubs Data Receiver. Это предоставляет управляемой учетной записи необходимые разрешения для отправки или получения сообщений во всех центрах событий в данном пространстве имен. Дополнительные сведения см. в статье Аутентификация приложения с помощью идентификатора Microsoft Entra для доступа к ресурсам Центров событий.
2. На вкладке "Члены" :
   1. При использовании управляемого удостоверения, назначаемого системой, для назначения доступа выберите "Пользователь", "Группа" или "Субъект-служба ", а затем выберите +Выберите участников и найдите имя расширения Azure IoT Operations Arc.
   2. При использовании управляемой идентичности, назначаемой пользователем, для назначения доступа выберите параметр "Управляемая идентичность", а затем выберите "Выбрать участников" и выполните поиск вашей управляемой идентичности, назначаемой пользователем, настроенной для облачных подключений.

Создание конечной точки для потока данных Центра событий Azure

После настройки пространства имен Центры событий Azure и концентратора событий можно создать конечную точку потока данных для пространства имен с поддержкой Kaf Центры событий Azure ka.

param aioInstanceName string = '<AIO_INSTANCE_NAME>'
param customLocationName string = '<CUSTOM_LOCATION_NAME>'
param endpointName string = '<ENDPOINT_NAME>'
param hostName string = '<NAMESPACE>.servicebus.windows.net:9093'

resource aioInstance 'Microsoft.IoTOperations/instances@2024-11-01' existing = {
  name: aioInstanceName
}
resource customLocation 'Microsoft.ExtendedLocation/customLocations@2021-08-31-preview' existing = {
  name: customLocationName
}
resource kafkaEndpoint 'Microsoft.IoTOperations/instances/dataflowEndpoints@2024-11-01' = {
  parent: aioInstance
  name: endpointName
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    endpointType: 'Kafka'
    kafkaSettings: {
      host: hostName
      authentication: {
        // See available authentication methods section for method types
        // method: <METHOD_TYPE>
      }
      tls: {
        mode: 'Enabled'
      }
    }
  }
}

az deployment group create --resource-group <RESOURCE_GROUP> --template-file <FILE>.bicep

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowEndpoint
metadata:
  name: <ENDPOINT_NAME>
  namespace: azure-iot-operations
spec:
  endpointType: Kafka
  kafkaSettings:
    host: <NAMESPACE>.servicebus.windows.net:9093
    authentication:
      # See available authentication methods section for method types
      # method: <METHOD_TYPE>
    tls:
      mode: Enabled

kubectl apply -f <FILE>.yaml

Используйте строку подключения для аутентификации в Центрах событий

kafkaSettings: {
  authentication: {
    method: 'Sasl'
    saslSettings: {
      saslType: 'Plain'
      secretRef: '<SECRET_NAME>'
    }
  }
  tls: {
    mode: 'Enabled'
  }
}

kubectl create secret generic <SECRET_NAME> -n azure-iot-operations \
  --from-literal=username='$ConnectionString' \
  --from-literal=password='Endpoint=sb://<NAMESPACE>.servicebus.windows.net/;SharedAccessKeyName=<KEY-NAME>;SharedAccessKey=<KEY>'

kafkaSettings:
  authentication:
    method: Sasl
    saslSettings:
      saslType: Plain
      secretRef: <SECRET_NAME>
  tls:
    mode: Enabled

Ограничения

Центры событий Azure не поддерживает все типы сжатия, поддерживаемые Kafka. В настоящее время в Центре событий Azure на уровнях «Премиум» и «Дедицированных» поддерживается только сжатие GZIP. Использование других типов сжатия может привести к ошибкам.

Пользовательские брокеры Kafka

Чтобы настроить конечную точку потока данных для Kafka-брокеров, не использующих Концентратор событий, установите сервер, TLS, проверку подлинности и другие параметры по мере необходимости.

resource kafkaEndpoint 'Microsoft.IoTOperations/instances/dataflowEndpoints@2024-11-01' = {
 parent: aioInstance
  name: '<ENDPOINT NAME>'
  extendedLocation: {
    name: '<CUSTOM_LOCATION_NAME>'
    type: 'CustomLocation'
  }
  properties: {
    endpointType: 'Kafka'
    host: '<KAFKA-HOST>:<PORT>'
    kafkaSettings: {
      authentication: {
        method: 'Sasl'
        saslSettings: {
            saslType: '<TYPE>'
            secretRef: '<SECRET_NAME>'
        }
      }
      tls: {
        mode: 'Enabled'
      }
    }
  }
}

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowEndpoint
metadata:
  name: <ENDPOINT NAME>
  namespace: azure-iot-operations
spec:
  endpointType: Kafka
  kafkaSettings:
    host: <KAFKA-HOST>:<PORT>
    authentication:
      method: Sasl
      saslSettings:
        saslType: <TYPE>
        secretRef: <SECRET_NAME>
    tls:
      mode: Enabled

Чтобы настроить параметры конечной точки, используйте следующие разделы для получения дополнительных сведений.

Доступные методы проверки подлинности

Для конечных точек потока данных брокера Kafka доступны следующие методы проверки подлинности.

Управляемое удостоверение, назначаемое системой

Перед настройкой конечной точки потока данных назначьте роль управляемому удостоверению Операций Интернета вещей Azure, который предоставляет разрешение на подключение к брокеру Kafka:

1. В портале Azure перейдите к экземпляру Azure IoT и выберите "Обзор".
2. Скопируйте имя расширения, указанного после расширения Azure IoT Operations Arc. Например, azure-iot-operations-xxxx7.
3. Перейдите к облачному ресурсу, которому необходимо предоставить разрешения. Например, перейдите в пространство имен Центров событий >Управление доступом (IAM)>, затем Добавить назначение роли.
4. На вкладке "Роль" выберите соответствующую роль.
5. На вкладке "Участники" для назначения доступа, выберите опцию "Пользователь, группа или служебный принципал", а затем нажмите опцию "+ Выбрать участников" и выполните поиск управляемого удостоверения Azure IoT Operations. Например, azure-iot-operations-xxxx7.

Затем настройте конечную точку потока данных с параметрами управляемой идентификации, назначенной системой.

kafkaSettings: {
  authentication: {
    method: 'SystemAssignedManagedIdentity'
    systemAssignedManagedIdentitySettings: {}
  }
}

kafkaSettings:
  authentication:
    method: SystemAssignedManagedIdentity
    systemAssignedManagedIdentitySettings:
      {}

Эта конфигурация создает управляемую идентификацию с аудиторией по умолчанию, совпадающей со значением хоста пространства имен Event Hubs в формате https://<NAMESPACE>.servicebus.windows.net. Однако если вам нужно переопределить аудиторию по умолчанию, можно задать audience для поля требуемое значение.

kafkaSettings: {
  authentication: {
    method: 'SystemAssignedManagedIdentity'
    systemAssignedManagedIdentitySettings: {
        audience: '<YOUR_AUDIENCE_OVERRIDE_VALUE>'
    }
  }
}

kafkaSettings:
  authentication:
    method: SystemAssignedManagedIdentity
    systemAssignedManagedIdentitySettings:
      audience: <YOUR_AUDIENCE_OVERRIDE_VALUE>

Управляемое удостоверение, назначаемое пользователем

Чтобы использовать назначаемое пользователем управляемое удостоверение для проверки подлинности, необходимо сначала развернуть операции Интернета вещей Azure с включенными безопасными параметрами. Затем необходимо настроить управляемое удостоверение, назначаемое пользователем, для облачных подключений. Дополнительные сведения см. в статье "Включение безопасных параметров в развертывании Операций Интернета вещей Azure".

Перед тем как настроить конечную точку потока данных, назначьте роль назначенной пользователем управляемой идентичности, которая предоставляет разрешение на подключение к брокеру Kafka:

1. В портал Azure перейдите к облачному ресурсу, которому необходимо предоставить разрешения. Например, перейдите к элементу >управления доступом (IAM) в пространстве имен Event Grid>Добавить назначение роли.
2. На вкладке "Роль" выберите соответствующую роль.
3. На вкладке Участники, для назначения доступа выберите параметр Управляемое удостоверение, затем нажмите + Выбрать участников и найдите ваше управляемое удостоверение, назначаемое пользователем.

Затем настройте конечную точку потока данных, используя параметры управляемого удостоверения, назначенного пользователем.

kafkaSettings: {
  authentication: {
    method: 'UserAssignedManagedIdentity'
    UserAssignedManagedIdentitySettings: {
      clientId: '<CLIENT_ID>'
      tenantId: '<TENANT_ID>'
      // Optional, defaults to https://<NAMESPACE>.servicebus.windows.net/.default
      // Matching the Event Hubs namespace you configured as host
      // scope: 'https://<SCOPE_URL>'
    }
  }
  ...
}

kafkaSettings:
  authentication:
    method: UserAssignedManagedIdentity
    userAssignedManagedIdentitySettings:
      clientId: <CLIENT_ID>
      tenantId: <TENANT_ID>
      # Optional, defaults to https://<NAMESPACE>.servicebus.windows.net/.default
      # Matching the Event Hubs namespace you configured as host
      # scope: https://<SCOPE_URL>

Здесь область применения — это аудитория управляемого удостоверения. Значение по умолчанию совпадает со значением узла пространства имен Центров событий в форме https://<NAMESPACE>.servicebus.windows.net. Однако если необходимо переопределить аудиторию по умолчанию, можно задать для поля области требуемое значение с помощью Bicep или Kubernetes.

SASL

Чтобы использовать SASL для проверки подлинности, укажите метод проверки подлинности SASL и настройте тип SASL и ссылку на секрет с именем секрета, содержащего маркер SASL.

kafkaSettings: {
  authentication: {
    method: 'Sasl' // Or ScramSha256, ScramSha512
    saslSettings: {
      saslType: 'Plain' // Or ScramSha256, ScramSha512
      secretRef: '<SECRET_NAME>'
    }
  }
}

kubectl create secret generic sasl-secret -n azure-iot-operations \
  --from-literal=token='<YOUR_SASL_TOKEN>'

kafkaSettings:
  authentication:
    method: Sasl
    saslSettings:
      saslType: Plain # Or ScramSha256, ScramSha512
      secretRef: <SECRET_NAME>

Поддерживаемые типы SASL:

• Plain
• ScramSha256
• ScramSha512

Секрет должен находиться в том же пространстве имен, что и конечная точка потока данных Kafka. Секрет должен иметь маркер SASL в качестве пары "ключ-значение".

Анонимные

Чтобы использовать анонимную проверку подлинности, обновите раздел проверки подлинности параметров Kafka, чтобы использовать анонимный метод.

kafkaSettings: {
  authentication: {
    method: 'Anonymous'
  }
}

kafkaSettings:
  authentication:
    method: Anonymous
    anonymousSettings:
      {}

Расширенные настройки

Дополнительные параметры можно задать для конечной точки потока данных Kafka, таких как TLS, доверенный сертификат ЦС, параметры обмена сообщениями Kafka, пакетная обработка и CloudEvents. Эти параметры можно задать на вкладке "Расширенный портал потока данных" или в ресурсе конечной точки потока данных.

// See sections below for more details
kafkaSettings: {
  consumerGroupId: '<ID>'
  compression: 'Gzip'
  copyMqttProperties: 'Disabled'
  kafkaAcks: 'All'
  partitionStrategy: 'Default'
  tls: {
    mode: 'Enabled'
    trustedCaCertificateConfigMapRef: '<YOUR_CA_CERTIFICATE>'
  }
  batching: {
    mode: 'Enabled'
    latencyMs: 1000
    maxMessages: 100
    maxBytes: 1024
  }
}

# See sections below for more details
kafkaSettings:
  consumerGroupId: <ID>
  compression: Gzip
  copyMqttProperties: Disabled
  kafkaAcks: All
  partitionStrategy: Default
  tls:
    mode: Enabled
    trustedCaCertificateConfigMapRef: <YOUR_CA_CERTIFICATE>
  batching:
    mode: Enabled
    latencyMs: 1000
    maxMessages: 100
    maxBytes: 1024

Параметры протокола TLS

Режим TLS

Чтобы включить или отключить TLS для конечной точки Kafka, обновите mode параметр в параметрах TLS.

kafkaSettings: {
  tls: {
    mode: 'Enabled' // Or Disabled
  }
}

kafkaSettings:
  tls:
    mode: Enabled # Or Disabled

Для режима TLS можно задать Enabled или Disabled. Если задан Enabledрежим, поток данных использует безопасное подключение к брокеру Kafka. Если задан Disabledрежим, поток данных использует небезопасное подключение к брокеру Kafka.

Сертификат доверенного удостоверяющего центра

Настройте доверенный сертификат ЦС для конечной точки Kafka, чтобы установить безопасное подключение к брокеру Kafka. Этот параметр важен, если брокер Kafka использует самоподписанный сертификат или сертификат, подписанный пользовательским ЦС, которому по умолчанию не доверяют.

kafkaSettings: {
  tls: {
    trustedCaCertificateConfigMapRef: '<YOUR_CA_CERTIFICATE>'
  }
}

kafkaSettings:
  tls:
    trustedCaCertificateConfigMapRef: <YOUR_CA_CERTIFICATE>

Этот "ConfigMap" должен содержать сертификат центра сертификации в формате PEM. ConfigMap должен находиться в том же пространстве имен, что и ресурс потока данных Kafka. Например:

kubectl create configmap client-ca-configmap --from-file root_ca.crt -n azure-iot-operations

Идентификатор группы потребителей

Идентификатор группы потребителей используется для идентификации группы потребителей, используемой потоком данных для чтения сообщений из раздела Kafka. Идентификатор группы потребителей должен быть уникальным в брокере Kafka.

kafkaSettings: {
  consumerGroupId: '<ID>'
}

spec:
  kafkaSettings:
    consumerGroupId: <ID>

Этот параметр действует только в том случае, если конечная точка используется в качестве источника (то есть поток данных является потребителем).

Сжатие

Поле сжатия позволяет сжимать сообщения, отправленные в разделы Kafka. Сжатие помогает сократить пропускную способность сети и пространство хранилища, необходимые для передачи данных. Однако сжатие также добавляет некоторые издержки и задержку в процесс. Поддерживаемые типы сжатия перечислены в следующей таблице.

Чтобы настроить сжатие, выполните приведенные действия.

kafkaSettings: {
  compression: 'Gzip' // Or Snappy, Lz4
}

kafkaSettings:
  compression: Gzip # Or Snappy, Lz4

Этот параметр действует только в том случае, если конечная точка используется в качестве назначения, где поток данных является производителем.

Пакетная обработка

Помимо сжатия, можно также настроить пакетную обработку сообщений перед отправкой в разделы Kafka. Пакетная обработка позволяет группировать несколько сообщений вместе и сжимать их в виде одной единицы, что может повысить эффективность сжатия и снизить нагрузку на сеть.

Например, если задержка равна 1000, maxMessages равно 100 и maxBytes равно 1,024, сообщения отправляются либо при наличии 100 сообщений в буфере, либо при наличии 1,024 байтов в буфере, либо как только истечет 1,000 миллисекунд с момента последней отправки.

Чтобы настроить пакетную обработку, выполните приведенные действия.

kafkaSettings: {
  batching: {
    mode: 'Enabled' // Or Disabled
    latencyMs: 1000
    maxMessages: 100
    maxBytes: 1024
  }
}

kafkaSettings:
  batching:
    mode: Enabled # Or Disabled
    latencyMs: 1000
    maxMessages: 100
    maxBytes: 1024

Этот параметр действует только в том случае, если конечная точка используется в качестве назначения, где поток данных является производителем.

Стратегия обработки разделов

Стратегия обработки секций определяет, как сообщения назначаются секциям Kafka при отправке их в разделы Kafka. Секции Kafka — это логические сегменты раздела Kafka, обеспечивающие параллельную обработку и отказоустойчивость. Каждое сообщение в теме Kafka имеет раздел и смещение, которые используются для идентификации и упорядочивания сообщений.

Этот параметр действует только в том случае, если конечная точка используется в качестве назначения, где поток данных является производителем.

По умолчанию поток данных назначает сообщения случайным секциям с помощью алгоритма циклического перебора. Однако можно использовать различные стратегии для назначения сообщений секциям на основе некоторых критериев, таких как имя раздела MQTT или свойство сообщения MQTT. Это поможет повысить балансировку нагрузки, локализацию данных или упорядочивание сообщений.

Например, если вы задали стратегию Property обработки секций и свойству device-idключа секции, сообщения с тем же свойством отправляются в ту же device-id секцию.

Чтобы настроить стратегию обработки секций, выполните следующие действия.

kafkaSettings: {
  partitionStrategy: 'Default' // Or Static, Topic, Property
  partitionKeyProperty: '<PROPERTY_NAME>'
}

kafkaSettings:
  partitionStrategy: Default # Or Static, Topic, Property
  partitionKeyProperty: <PROPERTY_NAME>

Благодарности Kafka

Подтверждения Kafka (acks) используются для управления устойчивостью и согласованностью сообщений, отправленных в разделы Kafka. Когда производитель отправляет сообщение в раздел Kafka, он может запрашивать различные уровни подтверждения от брокера Kafka, чтобы убедиться, что сообщение успешно записано в раздел и реплицируется в кластере Kafka.

Этот параметр действует только в том случае, если конечная точка используется в качестве назначения (т. е. поток данных является производителем).

Например, если задать подтверждение AllKafka, поток данных ожидает записи сообщения в раздел лидера и все секции подписчиков перед отправкой следующего сообщения.

Чтобы настроить подтверждения Kafka, выполните следующие действия.

kafkaSettings: {
  kafkaAcks: 'All' // Or None, One, Zero
}

kafkaSettings:
  kafkaAcks: All # Or None, One, Zero

Этот параметр действует только в том случае, если конечная точка используется в качестве назначения, где поток данных является производителем.

Копирование свойств MQTT

По умолчанию настройка копирования свойств MQTT включена. Эти свойства пользователя включают такие значения, как subject, которое хранит имя ресурса, отправляющего сообщение.

kafkaSettings: {
  copyMqttProperties: 'Enabled' // Or Disabled
}

kafkaSettings:
  copyMqttProperties: Enabled # Or Disabled

В следующих разделах описывается преобразование свойств MQTT в заголовки пользователей Kafka и наоборот, когда параметр включен.

Конечная точка Kafka — это точка назначения

Когда конечная точка Kafka является назначением потока данных, все определенные свойства спецификации MQTT версии 5 преобразуются в заголовки пользователей Kafka. Например, сообщение MQTT версии 5 с "Content Type", перенаправляемым в Kafka, преобразуется в пользовательский заголовок Kafka. Аналогичные правила применяются к другим встроенным свойствам MQTT, определенным в следующей таблице.

Пары значений ключей ключа свойства пользователя MQTT версии 5 напрямую преобразуются в заголовки пользователей Kafka. Если заголовок пользователя в сообщении имеет то же имя, что и встроенное свойство MQTT (например, заголовок с именем "Корреляционные данные"), то выбор между перенаправлением значения свойства спецификации MQTT версии 5 или пользовательского свойства не определен.

Потоки данных никогда не получают эти свойства от брокера MQTT. Таким образом, поток данных никогда не пересылает их:

• Псевдоним темы
• Идентификаторы подписки

Свойство "Интервал истечения срока действия сообщения"

Интервал истечения срока действия сообщения указывает, сколько времени сообщение может оставаться в брокере MQTT до его удаления.

Когда поток данных получает сообщение MQTT с указанным интервалом истечения срока действия сообщения, он:

• Записывает время получения сообщения.
• Перед отправкой сообщения в пункт назначения из начального времени истечения срока действия вычитается время, в течение которого сообщение находилось в очереди.
• Если сообщение не истекло (операция выше равно > 0), сообщение отправляется в место назначения и содержит обновленное время истечения срока действия сообщения.
• Если сообщение истекло (если результат операции выше это < = 0), сообщение не излучается целевым объектом.

Примеры:

• Поток данных получает сообщение MQTT с интервалом истечения срока действия сообщения = 3600 секунд. Соответствующий пункт назначения временно отключен, но может повторно подключиться. 1 000 секунд проходит, прежде чем это сообщение MQTT отправляется на целевой узел. В этом случае для сообщения назначения установлен интервал истечения срока действия 2600 (3600 - 1000) секунд.
• Поток данных получает сообщение MQTT с интервалом истечения срока действия сообщения = 3600 секунд. Соответствующий адрес временно отключён, но может подключиться снова. Однако в этом случае для повторного подключения требуется 4000 секунд. Срок действия сообщения истек, и поток данных не перенаправит это сообщение в место назначения.

Конечная точка Kafka — это источник потока данных

Когда конечная точка Kafka является источником потока данных, заголовки пользователей Kafka преобразуются в свойства MQTT версии 5. В следующей таблице описывается преобразование заголовков пользователей Kafka в свойства MQTT версии 5.

Пары ключей и значений пользовательских заголовков Kafka, при условии, что они все закодированы в UTF-8, напрямую преобразуются в пользовательские свойства ключей и значений MQTT.

UTF-8 / двоичные несоответствия

MQTT версии 5 поддерживает только свойства на основе UTF-8. Если поток данных получает сообщение Kafka, содержащее один или несколько заголовков, отличных от UTF-8, поток данных:

• Удалите проблемное свойство или свойства.
• Перенаправите остальное сообщение в соответствии с предыдущими правилами.

Приложения, для которых требуется двоичная передача в заголовках источника Kafka => свойства MQTT Target должны сначала кодировать их в кодировке UTF-8, например с помощью Base64.

>=64 КБ несоответствия свойств

Свойства MQTT версии 5 должны быть меньше 64 КБ. Если поток данных получает сообщение Kafka, содержащее один или несколько заголовков размером >= 64 КБ, поток данных будет:

• Удалите проблемное свойство или свойства.
• Перенаправите остальное сообщение в соответствии с предыдущими правилами.

Преобразование свойств при использовании Центров событий и поставщиков, использующих AMQP

Если ваш клиент пересылает сообщения на конечную точку источника потока данных Kafka, выполняющую одно из следующих действий:

• Отправка сообщений в Центры событий с помощью клиентских библиотек, таких как Azure.Messaging.EventHubs
• Использование AMQP напрямую

Существуют нюансы перевода свойств, которые следует учитывать.

Необходимо выполнить одно из следующих действий:

• Избегайте отправки свойств
• Если необходимо отправить свойства, отправьте значения, закодированные как UTF-8.

Когда Центр событий преобразовывает свойства из AMQP в Kafka, он включает базовые типы, закодированные AMQP в своем сообщении. Дополнительные сведения о поведении см. в разделе "Обмен событиями между потребителями и производителями" с помощью различных протоколов.

В следующем примере кода, когда конечная точка потока данных получает значение "foo":"bar", оно получает свойство как <0xA1 0x03 "bar">.

using global::Azure.Messaging.EventHubs;
using global::Azure.Messaging.EventHubs.Producer;

var propertyEventBody = new BinaryData("payload");

var propertyEventData = new EventData(propertyEventBody)
{
  Properties =
  {
    {"foo", "bar"},
  }
};

var propertyEventAdded = eventBatch.TryAdd(propertyEventData);
await producerClient.SendAsync(eventBatch);

Конечная точка потока данных не может перенаправить свойство <0xA1 0x03 "bar"> нагрузки в сообщение MQTT, так как данные не являются в формате UTF-8. Однако если указать строку UTF-8, конечная точка потока данных преобразует строку перед отправкой в MQTT. Если используется строка UTF-8, сообщение MQTT будет иметь "foo":"bar" в качестве свойств пользователя.

Переводятся только заголовки UTF-8. Например, учитывая следующий сценарий, в котором свойство задано как float:

Properties = 
{
  {"float-value", 11.9 },
}

Конечная точка потока данных удаляет пакеты, содержащие "float-value" поле.

Не все свойства данных событий, например, propertyEventData.correlationId, перенаправляются. Дополнительные сведения см. в разделе "Свойства пользователя события",

CloudEvents

CloudEvents — это способ описания данных событий распространенным способом. Параметры CloudEvents используются для отправки или получения сообщений в формате CloudEvents. Вы можете использовать CloudEvents для архитектур на основе событий, где разные службы должны взаимодействовать друг с другом в одном или разных поставщиках облачных служб.

Параметры: CloudEventAttributes — это Propagate или CreateOrRemap.

kafkaSettings: {
  cloudEventAttributes: 'Propagate' // Or CreateOrRemap
}

kafkaSettings:
  cloudEventAttributes: Propagate # Or CreateOrRemap

В следующих разделах описывается, как свойства CloudEvent распространяются, создаются или изменяются и переназначаются.

Параметр распространения

Свойства CloudEvent передаются для сообщений, содержащих необходимые свойства. Если сообщение не содержит необходимых свойств, сообщение передается как есть. Если необходимые свойства присутствуют, ce_ префикс добавляется в имя свойства CloudEvent.

Параметр CreateOrRemap

Свойства CloudEvent передаются для сообщений, содержащих необходимые свойства. Если сообщение не содержит обязательных свойств, создаются свойства.

Следующие шаги

Дополнительные сведения о потоках данных см. в статье "Создание потока данных".