apache-kafka - 使用 Debezium 提取 key 的转换中不存在字段

Question

我正在尝试创建一个 Debezium MySQL 连接器，并进行转换以提取 key 。

关键改造前:

create source connector mysql with(
    "connector.class" = 'io.debezium.connector.mysql.MySqlConnector',
    "database.hostname" = 'mysql',
    "tasks.max" = '1',
    "database.port" = '3306',
    "database.user" = 'debezium',
    "database.password" = 'dbz',
    "database.server.id" = '42',
    "database.server.name" = 'before',
    "table.whitelist" = 'deepprices.deepprices',
    "database.history.kafka.bootstrap.servers" = 'kafka:29092',
    "database.history.kafka.topic" = 'dbz.deepprices',
    "include.schema.changes" = 'true',
    "transforms" = 'unwrap',
    "transforms.unwrap.type" = 'io.debezium.transforms.UnwrapFromEnvelope');

主题结果是:

> rowtime: 2020/05/20 16:47:23.354 Z, key: [St@5778462697648631933/8247607644536792125], value: {"id": "P195910", "price": "1511.64"}

当 key.converter 设置为 JSON 时，Key 变为 {"id": "P195910"}
所以，我想从键中提取 id 并将其设为字符串键:

预期成绩 :

rowtime: 2020/05/20 16:47:23.354 Z, 
key: 'P195910', 
value: {"id": "P195910", "price": "1511.64"}   

尝试使用 ExtractField 进行转换时或 ValueToKey我得到:

DataException: Field does not exist: id:

我尝试使用包含 ValueToKey 的指令:

create source connector mysql with(
    "connector.class" = 'io.debezium.connector.mysql.MySqlConnector',
    "database.hostname" = 'mysql',
    "tasks.max" = '1',
    "database.port" = '3306',
    "database.user" = 'debezium',
    "database.password" = 'dbz',
    "database.server.id" = '42',
    "database.server.name" = 'after',
    "table.whitelist" = 'deepprices.deepprices',
    "database.history.kafka.bootstrap.servers" = 'kafka:29092',
    "database.history.kafka.topic" = 'dbz.deepprices',
    "include.schema.changes" = 'true',
    "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',
    "transforms" = 'unwrap,createkey',
    "transforms.unwrap.type" = 'io.debezium.transforms.UnwrapFromEnvelope',
    "transforms.createkey.type" = 'org.apache.kafka.connect.transforms.ValueToKey',
    "transforms.createkey.fields" = 'id'
    );

导致我的 出现以下错误Kafka 连接日志 :

Caused by: org.apache.kafka.connect.errors.DataException: Field does not exist: id
        at org.apache.kafka.connect.transforms.ValueToKey.applyWithSchema(ValueToKey.java:89)
        at org.apache.kafka.connect.transforms.ValueToKey.apply(ValueToKey.java:67)

Answer 1

从 UnwrapFromEnvelope 更改转换类型至 ExtractNewRecordState ，解决了 Debezium MySQL CDC 连接器的问题，版本 1.1.0 .

transforms.unwrap.type" = 'io.debezium.transforms.ExtractNewRecordState'