13.5. 过滤 Debezium 更改事件记录

为安全起见，Debezium 连接器存档中没有包括过滤 SMT。相反，它会在单独的工件 debezium-scripting-2.3.7.Final.tar.gz 中提供。

如果您通过从 Dockerfile 构建自定义 Kafka Connect 容器镜像来部署 Debezium 连接器，以使用过滤器 SMT，您必须明确下载 SMT 归档，并将文件与连接器插件一起部署。当使用 AMQ Streams 部署连接器时，它可以根据您在 Kafka Connect 自定义资源中指定的配置参数自动下载所需的工件。重要信息：在 Kafka Connect 实例中存在过滤器 SMT 后，允许向实例添加连接器的任何用户都可以运行脚本表达式。为确保脚本表达式只能由授权用户运行，请务必在添加过滤器 SMT 前保护 Kafka Connect 实例及其配置接口。

如果您从 Dockerfile 构建 Kafka Connect 容器镜像，则应用以下步骤。如果使用 AMQ Streams 创建 Kafka Connect 镜像，请按照您的连接器部署主题中的说明进行操作。

Groovy 语言在 classpath 中需要以下库：

JavaScript 语言在 classpath 中需要以下库：

您可以在 Debezium 连接器的 Kafka Connect 配置中配置过滤器转换。在配置中，您可以通过定义基于业务规则的过滤器条件来指定您感兴趣的事件。当过滤 SMT 处理事件流时，它会根据配置的过滤器条件评估每个事件。只有满足过滤器条件条件的事件才会传递给代理。

要配置 Debezium 连接器来过滤更改事件记录，请在 Debezium 连接器的 Kafka Connect 配置中配置 Filter SMT。配置过滤器 SMT 要求您指定一个定义过滤条件的正则表达式。

例如，您可以在连接器配置中添加以下配置。

...
transforms=filter
transforms.filter.type=io.debezium.transforms.Filter
transforms.filter.language=jsr223.groovy
transforms.filter.condition=value.op == 'u' && value.before.id == 2
...

...
transforms=filter
transforms.filter.type=io.debezium.transforms.Filter
transforms.filter.language=jsr223.groovy
transforms.filter.condition=value.op == 'u' && value.before.id == 2
...

前面的示例指定了 Groovy 表达式语言的使用。正则表达式 value.op == 'u' && value.before.id == 2 会删除所有消息，但代表更新(u)的 id 值除外，其 id 值等于 2。

Debezium 将某些变量绑定到过滤器 SMT 的评估上下文中。在创建表达式来指定过滤器条件时，您可以使用 Debezium 绑定到评估上下文的变量。通过绑定变量，Debebe 可让 SMT 查找和解释它们的值，因为它评估表达式中的条件。

下表列出了 Debezium 绑定到过滤器 SMT 的评估上下文的变量：

表达式可以在其变量上调用任意方法。表达式应该解析为布尔值，它决定了 SMT 分布消息的方式。当表达式中的过滤器条件评估为 true 时，消息会被保留。当过滤器条件评估为 false 时，消息将被删除。

表达式不应产生任何副作用。也就是说，它们不应修改它们通过的任何变量。

除了 Debezium 连接器在数据库更改时发出的更改事件消息外，连接器还会发出其他类型的信息，包括心跳消息，以及有关 schema 更改和事务的元数据消息。由于这些其他消息的结构与 SMT 设计的更改事件消息的结构不同，因此最好将连接器配置为有选择地应用 SMT，以便它只处理预期的数据更改消息。您可以使用以下任一方法配置连接器来有选择地应用 SMT：

表达过滤条件的方式取决于您使用的脚本语言。

例如，如 基本配置示例 所示，当您使用 Groovy 作为表达式语言时，以下表达式会删除所有消息，但更新记录除外，其 id 值设置为 2：

value.op == 'u' && value.before.id == 2

value.op == 'u' && value.before.id == 2

其他语言使用不同的方法来表达相同的条件。

value.get('op') == 'u' && value.get('before').get('id') == 2

value.get('op') == 'u' && value.get('before').get('id') == 2

value.op == 'u' && value.before.id == 2

value.op == 'u' && value.before.id == 2

下表列出了可与过滤器 SMT 一起使用的配置选项。