提取和转换jdbc接收器连接器的kafka消息特定字段

Question

我有一个kafka主题，它是使用Debezium mysql source connector从mysql数据库获取数据，下面是其中一条消息的格式：

{
    "Message": {
        "schema": {
            "type": "struct",
            "fields": [
              ...
            ],
            "optional": true,
            "name": "mysql-server-1.inventory.somename"
        },
        "payload": {
            "op": "u",
            "ts_ms": 1465491411815,
            "before": {
                "id": 1004,
                "first_name": "Anne",
                "last_name": "Doof",
                "email": "[e-mail]"
            },
            "after": {
                "id": 1004,
                "first_name": "Anne",
                "last_name": "Marry",
                "email": "[e-mail]"
            },
            "source": {
                "db": "inventory",
                "table": "customers",
                ...
                "query": "Update customers set last_name = 'Marry' where id = 1004"
            }
        }
    }
}

我想使用jdbc接收器连接器将ts_ms,before,after和id（从对象/行）列推送到另一个数据库中,表模式为(id,before（text）,after（text）,timestamp),对kafka来说,我不知道：

>

我如何才能只提取这些字段，从消息中推送而忽略其他字段？

如何将before、after字段转换为字符串/序列化格式？

如何从对象提取id？（如果是insert操作，则before为空，如果是delete操作，则after为空）

对于上面的消息,接收器目标表的末尾应该有如下所示的数据：

id:     1004
before: '{"id":1004,"first_name":"Anne","last_name":"Doof","email":"[e-mail]"}'
after:  '{"id":1004,"first_name":"Anne","last_name":"Marry","email":"[e-mail]"}'
timestamp: 1465491411815

Answer 1

您可以为您的json有效负载创建一个DTO（您从kafka主题中获得的Java对象），利用这个联机转换器可以帮助您将json转换为Java对象。[http://pojo.sodhanalibrary.com/][1]

一旦从kafka主题接收到消息，就可以使用objectmapper转换json并将其映射到适当的DTO对象。您可以通过调用getId()、getBefore()等来使用该对象提取所需的字段,

下面是一些参考代码，可以帮助您理解：

    @KafkaListener(topics = "test")
        public void listen(String payload)  {

            logger.info("Message Received from Kafka topic: {}", payload);

            ObjectMapper objectMapper = new ObjectMapper();
            objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);

            DTOObject dtoObject = objectMapper.readValue(payload,DTOObject.class);

                logger.info("After Convertion: {}", objectMapper.writeValueAsString(dtoObject));

                logger.info("Get Before:{}", dtoObject.getId());



        }

Answer 2

您可以使用卡夫卡连接转换链，就像这个解决方案。