如何配置 Debezium 以使用特定列作为 Kafka 消息键？

【问题标题】：How to configure Debezium to use specific column as Kafka message key?如何配置 Debezium 以使用特定列作为 Kafka 消息键？
【发布时间】：2020-06-29 00:30:11
【问题描述】：

默认情况下，Debezium 使用表的主键作为消息键。例如，如果你有一张桌子

create table users
(
    id            bigint auto_increment primary key,
    department_id bigint
);

有数据

+----+----------------+
| id | department_id  |
+----+----------------+
|  5 |              1 |
|  6 |              1 |
|  7 |              2 |
+----+----------------+

Debezium 将产生以下 Kafka 消息：

Key: {"id": 5} Value: {"id": 5, "department_id": 1}
Key: {"id": 6} Value: {"id": 6, "department_id": 1}
Key: {"id": 7} Value: {"id": 7, "department_id": 2}

问题是如何配置 Debezium 以使用 department_id 或任何其他列作为 Kafka 消息密钥？

【问题讨论】：

标签： mysql apache-kafka-connect debezium

【解决方案1】：

为此有message.key.columns 参数。在连接器的配置中，您应该这样设置：

{
  "name": "my-connector",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "tasks.max": "1",
    "database.hostname": "mysql",
    "database.port": "3306",
    "database.whitelist": "my_database",
    ...
    "message.key.columns": "my_database.users:department_id"
  }
}

所有关系型 Debezium 连接器都支持此参数。

【讨论】：

请记住，此时不应压缩您的主题，因为您不会保留每个用户的最后一个事件，而是压缩后每个部门的最后一个事件。此外，当更改用户的部门时，这些事件可能会在不同的分区中结束，因此您不会有任何保证的顺序。