Kafka SMT ValueToKey：如何将多个字段值拼接为消息键？

我明白你的困扰——ValueToKey默认会把你指定的字段打包成一个Struct类型的键（比如Struct{field1: "val1", field2: "val2", field3: "val3"}），但你想要的是直接把三个字段的值拼接成一个字符串（比如val1val2val3或者val1-val2-val3）。下面给你两种实用的解决方案，根据你的场景选择：

方案一：在Oracle查询层面拼接字段（最简单） #

如果你能修改JDBC Connector的查询语句，直接在Oracle里把三个字段拼接成一个新字段，再用ValueToKey提取这个字段作为键，这是最直接的方式，不需要额外的SMT插件。

配置示例：

# JDBC Source Connector基础配置
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
connection.url=jdbc:oracle:thin:@//your-oracle-host:1521/your-db
connection.user=your-user
connection.password=your-password
mode=incrementing
incrementing.column.name=your-id-column

# 自定义查询，拼接三个字段为composite_key
query=SELECT field1, field2, field3, NVL(field1,'') || NVL(field2,'') || NVL(field3,'') AS composite_key FROM your_table

# 使用ValueToKey提取拼接后的字段作为键
transforms=ValueToKey
transforms.ValueToKey.type=org.apache.kafka.connect.transforms.ValueToKey
transforms.ValueToKey.fields=composite_key

注意：用||运算符拼接更灵活，搭配NVL可以处理字段为空的情况，避免拼接结果变成空值。如果是Oracle 12c+，也可以用CONCAT_WS函数指定分隔符，比如CONCAT_WS('-', NVL(field1,''), NVL(field2,''), NVL(field3,''))。

方案二：使用SMT插件拼接字段（无需修改查询） #

如果不想修改查询语句，可以使用Confluent Hub提供的kafka-connect-string-transforms插件，它包含专门的字符串拼接SMT。

步骤1：安装插件 #

在你的Connector节点上执行：

confluent-hub install confluentinc/kafka-connect-string-transforms:latest

步骤2：配置Connector #

# JDBC Source Connector基础配置（省略重复部分）
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
# ... 其他基础配置 ...

# 先通过ValueToKey把三个字段提取为键的Struct，再拼接成字符串
transforms=ValueToKey, ConcatKeys

# 第一步：提取三个字段到键的Struct中
transforms.ValueToKey.type=org.apache.kafka.connect.transforms.ValueToKey
transforms.ValueToKey.fields=field1,field2,field3

# 第二步：把Struct中的三个字段值拼接成字符串
transforms.ConcatKeys.type=io.confluent.connect.transforms.Concat$Key
transforms.ConcatKeys.fields=field1,field2,field3
# 可选：设置分隔符，比如用"-"分隔，默认是空字符串直接拼接
transforms.ConcatKeys.separator=-

这样配置后，消息键就会变成你想要的拼接字符串了。

为什么ValueToKey不符合预期？ #

再补充一下：ValueToKey的核心作用是从消息值中提取指定字段，包装成一个Struct类型的键，它本身并不做字符串拼接。所以你需要额外的步骤（要么在查询里预处理，要么用拼接SMT）把Struct转换成拼接后的字符串。

内容的提问来源于stack exchange，提问作者Alex