具有动态值的 Kafka Connect InsertField 转换答案

【问题标题】：Kafka Connect InsertField transform with dynamic values具有动态值的 Kafka Connect InsertField 转换
【发布时间】：2020-09-27 16:47:41
【问题描述】：

我遇到了需要向 Kafka Connect 记录插入新字段的情况，但 InsertField 转换似乎仅限于静态值。

https://docs.confluent.io/current/connect/transforms/insertfield.html

有没有办法根据记录中的其他字段添加动态值？

我需要这个的原因是因为我使用 JDBC Source 和 Sink 连接器在两个数据库之间传输数据。在水槽方面，我正在做 upserts。因此，我需要以下字段：

        "insert.mode": "upsert",
        "pk.mode": "record_value",
        "pk.fields": "TABLE_ID",

这很好用，但它让我每个表都有一个连接器文件，因为在我的源数据库中，所有表都具有 tableName_ID 形式的主键。所以 pk.fields 总是会有所不同。

我想在我的接收器数据库中添加一个我的应用程序不知道的新字段，但 Kafka Connect 将用于更新主键。它将被称为kafka_id 之类的名称，并且对于每个表都是相同的。我想在我的源配置中添加这个字段，然后在我的接收器中更新 pk.fields。有什么办法可以做到这一点吗？我需要编写自定义转换吗？谢谢！

【问题讨论】：

也许这种转换jcustenborder.github.io/kafka-connect-documentation/projects/… 可能有用
好建议，我实际上正在研究那个。对我来说不幸的是， field.pattern 必须基于我认为我无法从那里访问的主题/表名称？我不认为我可以做类似 field.pattern:"^${topic}".
我认为你可以使用模式.*_id。另一种选择是您自己的 SMT：ConnectRecord 有关于其主题的信息 (kafka.apache.org/23/javadoc/org/apache/kafka/connect/connector/…)。
啊，好的，我明白你现在在说什么了。这是另一个好主意，但不幸的是，我的行中可能（而且几乎肯定会）有其他字段与相同的模式 SOMETHING_ID 匹配。主要是外键。这样做最终会在 json "transforms.patternRename.field.pattern" : "^.*_ID", "transforms.patternRename.field.replacement" : "id" 中产生多个具有相同名称的字段。我认为我最好的选择可能是我自己的 SMT。感谢您的链接
不客气！

标签： apache-kafka transform apache-kafka-connect upsert

【解决方案1】：

在您的情况下，最好的选择是您自己的 Kafka Connect SMT：您可以使用来自 ConnectRecord 的主题信息。

Example 的 SMT 来自 How to Use Single Message Transforms in Kafka Connect 文章。

【讨论】：