如何为IoT中心的每个新生产者添加新的Kafka主题？

我正在研究Azure云解决方案.我正在使用连接到Kafka的IoT中心来处理来自各种IoT设备的数据.
我面临的是,来自多个设备的所有数据都存储在同一主题中.但是,我想将连接到IoT中心的每个设备的数据处理到Kafka中的特定主题(每个设备都有自己的Kafka主题)

Toketi“ Azure IoT中心的Kafka Connect源连接器”提供以下配置文件(边缘节点)

connector.class=com.microsoft.azure.iot.kafka.connect.source.IotHubSourceConnector
name=AzureIotHubConnector
tasks.max=1
Kafka.Topic=IotTopic
IotHub.EventHubCompatibleName=iothub-toketi
IotHub.EventHubCompatibleEndpoint=sb://iothub-001.servicebus.windows.net/
IotHub.AccessKeyName=service
IotHub.AccessKeyValue=4KsdfiB9J899a+N3iwerjKwzeqbZUj1K//KKj1ye9i3=
IotHub.ConsumerGroup=$Default
IotHub.Partitions=4
IotHub.StartTime=2016-11-28T00:00:00Z
IotHub.Offsets=
BatchSize=100
ReceiveTimeout=60

它适用于一个主题,用于存储来自多个设备的所有数据,但是我希望对来自设备的数据进行隔离

任何解决方案或想法！

谢谢

解决方法:

解决方案之一是使用SMT(单消息转换).

源连接器流程包含几个步骤：

>以List< SourceRecord>从外部源轮询数据；
>使用定义的SMT转换每个消息(SourceRecord)(如果未定义转换,则可以跳过
>将SourceRecord的键和值转换为字节数组.
>通过KafkaProducer向Kafka发送消息

Kafka Connect根据SourceRecord :: topic字段确定向哪个主题发送消息.使用SMT,您可以设置适当的主题值.

纯Apache Kafka Connect没有这种转换.
如果您正在使用Confluent Platform,则可以使用其他一些转换.
要提取主题名称,可以使用ExtractTopic.它具有属性,即字段

有关SMT整个概念的更多信息,请参见Apache Kafka web page或Confluent web page

标签：apache-kafka-connect,azure,apache-kafka,azure-iot-hub,c
来源： https://codeday.me/bug/20191210/2105093.html