如何通过Debezium Connect反序列化Kafka消息流中的几何字段

Question

我有一个PostGIS+Debezium/Kafka+Debezium/Connect设置，它将更改从一个数据库流式传输到另一个数据库。我一直在通过Kowl观看消息，一切都在相应地进行。

我的问题取决于我读卡夫卡主题的信息，尤其是几何专栏。

这是我的卡夫卡信息：

{
"schema":{
"type":"struct"
"fields":[...]
"optional":false
"name":"ecotx_geometry_kafka.ecotx_geometry_impo..."
}
"payload":{
"before":NULL
"after":{
"id":"d6ad5eb9-d1cb-4f91-949c-7cfb59fb07e2"
"type":"MultiPolygon"
"layer_id":"244458fa-e6e0-4c6c-a7e1-5bf0afce2fb8"
"geometry":{
"wkb":"AQYAACBqCAAAAQAAAAEDAAAAAQAAAAUAAABwQfUo..."
"srid":2154
}
"custom_style":NULL
"style_id":"default_layer_style"
}
"source":{...}
"op":"c"
"ts_ms":1618854994546
"transaction":NULL
}
}

可以看出，WKB的信息有点像"；AQAAAAA&"；，尽管在我的数据库中插入的信息是"；01060000208A7A0000000000000"；或"；LINESTRING(0,10("；。

我不知道如何在我的Consumer应用程序(Kotlin/Java(中将其解析/转换为ByteArray或Geometry，以便在GeoTools中进一步使用。

我不知道我是否错过了一个能够翻译这些信息的导入。

我对人们发布json消息有几个问题，每一条有geom字段(流式传输w/Debezium(的消息都改成了这个"；AAAQQQAAAA"；。

话虽如此，我如何解析/解码/翻译成GeoTools可以使用的东西？

谢谢。

@更新

附加信息：

插入后，当我分析插槽更改时(使用pg_logical_slot_get_changes函数查询数据库(，我可以在WKB:中看到我的更改

{"change":[{"kind":"insert","schema":"ecotx_geometry_import","table":"geometry_data","columnnames":["id","type","layer_id","geometry","custom_style","style_id"],"columntypes":["uuid","character varying(255)","uuid","geometry","character varying","character varying"],"columnvalues":["469f5aed-a2ea-48ca-b7d2-fe6e54b27053","MultiPolygon","244458fa-e6e0-4c6c-a7e1-5bf0afce2fb8","01060000206A08000001000000010300000001000000050000007041F528CB332C413B509BE9710A594134371E05CC332C4111F40B87720A594147E56566CD332C4198DF5D7F720A594185EF3C8ACC332C41C03BEDE1710A59417041F528CB332C413B509BE9710A5941",null,"default_layer_style"]}]}

这在消费者应用程序中很有用，它肯定依赖于Kafka Message内容本身，只是不确定是谁在改变这个价值，如果是Kafka还是DBZ/Connect。

Answer 1

我认为这只是PostGIS和JSON中表示二进制列的一种不同方式。WKB是一个二进制字段，这意味着它包含具有任意值的字节，其中许多字节没有相应的可打印字符。PostGIS使用HEX编码打印出来，因此它看起来像"01060000208A7A…"-十六进制数字，但内部只是字节。Kafka的JSON对完全相同的二进制消息使用BASE64编码。

让我们用字符串的前缀进行测试

select to_base64(from_hex('01060000206A080000010000000103000000010000000500'))
AQYAACBqCAAAAQAAAAEDAAAAAQAAAAUA