我正在尝试使用google pubsub源连接器从我的google云获取数据到kafka。我确实得到了数据,但是消息是以字节的形式出现的。我在这里提到过,如前所述,我使用了json转换器来更改它。
这是我的连接器代码部分:
name=CPSSourceConnector
connector.class=com.google.pubsub.kafka.source.CloudPubSubSourceConnector
tasks.max=10
kafka.topic=test-topic
kafka.topic.replication.factor=1
kafka.key.attribute=message
key.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=true
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=true
cps.subscription=test-sub
cps.project=sensor-alpha
这就是我在Kafka中得到的:
{
"schema":{
"type":"struct",
"fields":[
{
"type":"bytes",
"optional":false,
"field":"message"
},
{
"type":"string",
"optional":false,
"field":"subFolder"
},
{
"type":"string",
"optional":false,
"field":"deviceId"
},
{
"type":"string",
"optional":false,
"field":"deviceRegistryLocation"
},
{
"type":"string",
"optional":false,
"field":"projectId"
},
{
"type":"string",
"optional":false,
"field":"deviceNumId"
},
{
"type":"string",
"optional":false,
"field":"deviceRegistryId"
}
],
"optional":false
},
"payload":{
"message":"eyJzZW5zb3JfaWQiOiAiYmEwMGQyNjNiNzRiMzBhMGFjM2EzMDlkZWZjZjM0ODMtMzAyIiwgInRfY2Vsc2l1cyI6IDEwLCAicmVnaXN0cnlfaWQiOiAiYmFsZW5hLXJlZ2lzdHJ5IiwgInByZXNzdXJlIjogMTAsICJ0aW1lc3RhbXAiOiAxNTk4NDM2NTk3LjQxNTEwNDYsICJkZXZpY2VfaWQiOiAiYmEwMGQyNjNiNzRiMzBhMGFjM2EzMDlkZWZjZjM0ODMiLCAic3RyaW5nX2JhdHRlcnkiOiAiYmF0dGVyeV9ub3JtYWwiLCAic3RyaW5nX2luZmxhdGUiOiAidGlyZV9vdmVyX2luZmxhdGVkIn0=",
"subFolder":"",
"deviceId":"deviceid",
"deviceRegistryLocation":"region_value",
"projectId":"projectid",
"deviceNumId":"device_num_value",
"deviceRegistryId":"registryid"
}
}
即使在提供了连接器之后,我也会以字节的形式获得消息的细节。我还需要做些什么把它转换成json格式吗?
1条答案
按热度按时间ibps3vxo1#
云pub/sub-kafka连接器不检查或转换它接收到的消息中的数据;它只是以字节的形式传递数据字段,这是pubsubmessage中字段的类型。目前无法让连接器本身读取消息的内容并将其转换为json。