Kafka Connect迁移SQL Server数据时时间戳转Epoch毫秒问题咨询

核心原因说明 #

你之前配置TimestampConverter SMT不生效，大概率是漏了两处关键配置：一是Debezium SQL Server连接器默认会直接把日期时间字段输出为Long型的Epoch毫秒值，输入到SMT的字段已经不是时间类型，导致SMT匹配不到；二是JSON转换器默认会把Connect的时间类型转为Epoch值，就算SMT转换成功也会被二次转换。

完整配置方案 #

1. 调整连接器时间精度配置 #

首先在SQL Server连接器配置中添加如下参数，让连接器输出标准的Connect时间类型而非直接转Long：

time.precision.mode=connect

2. 配置正确的TimestampConverter SMT #

按照你的格式需求配置SMT，示例配置如下：

transforms=timestampConverter
# 如果要处理消息Key中的时间字段，把下方的$Value改成$Key
transforms.timestampConverter.type=org.apache.kafka.connect.transforms.TimestampConverter$Value
transforms.timestampConverter.target.type=string
# 自定义时间格式，可按需求调整
transforms.timestampConverter.format=yyyy-MM-dd HH:mm:ss.SSS
# 如需仅转换指定字段，添加下方配置，值为你的日期时间字段名，不配置则全局处理所有时间类型
# transforms.timestampConverter.field=create_time

3. 调整JSON转换器配置 #

禁用JSON转换器默认的时间转Epoch逻辑，配置如下：

value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false
value.converter.write.dates.as.iso8601=true
value.converter.write.timestamps.as.iso8601=true
value.converter.write.datetimes.as.iso8601=true
# 如Key中也存在日期时间字段，同步添加Key转换器的对应配置
key.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=false
key.converter.write.dates.as.iso8601=true
key.converter.write.timestamps.as.iso8601=true
key.converter.write.datetimes.as.iso8601=true

注意事项 #

• SMT的配置顺序要放在其他字段转换逻辑之前，避免字段类型被提前修改导致匹配失败
• 重启连接器后生效，可先测试单条数据验证输出格式是否符合预期

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