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

Question

我有一个 kafka 主题，它使用 Debezium mysql 源连接器从 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": "annek@noanswer.org"
            },
            "after": {
                "id": 1004,
                "first_name": "Anne",
                "last_name": "Marry",
                "email": "annek@noanswer.org"
            },
            "source": {
                "db": "inventory",
                "table": "customers",
                ...
                "query": "Update customers set last_name = 'Marry' where id = 1004"
            }
        }
    }
}

我想使用 jdbc sink 连接器将 ts_ms, before, after 和 id（从对象/行）列推送到另一个数据库，表架构为 (id,before(text),after(text),timestamp)，对于 kafka 的新手无法弄清楚：

• 如何仅从消息中提取这些字段以推送并忽略其他字段？

• 如何将字段前后转换为字符串/序列化格式？

• 如何从对象中提取id？ （插入操作，before为null，delete，after为null）

对于上面的消息，sink目标表最后应该有如下数据：

id:     1004
before: '{"id":1004,"first_name":"Anne","last_name":"Doof","email":"annek@noanswer.org"}'
after:  '{"id":1004,"first_name":"Anne","last_name":"Marry","email":"annek@noanswer.org"}'
timestamp: 1465491411815

Answer 1

您可以使用Kafka Connect Transformations 的链，例如solution。

Answer 2

您可以创建一个 DTO（您从 kafka 主题中获得的 json 有效负载的 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());



        }