Kafka Connect CDC同步Couchbase：单字段更新与时间格式配置

我们当前基于CDC链路实现MySQL到Couchbase的数据同步，全链路环境已部署就绪，Debezium源端已可正常向Kafka生产符合预期的变更消息，针对现存两个问题的解决方案如下：

问题1：Debezium将DATETIME类型序列化为Unix时间戳，需输出YYYY-mm-dd HH:MM:SS格式字符串 #

解决方案 #

直接在现有Debezium MySQL源连接器（our-connector）的config配置块中添加以下时间格式配置即可，无需额外开发转换逻辑：

# 时间精度模式设置为connect，关闭默认返回Unix时间戳的逻辑
time.precision.mode: "connect"
# 指定DATETIME类型的输出格式，严格匹配YYYY-MM-dd HH:mm:ss（注意格式符大小写：MM代表月份，mm代表分钟，不要写错）
datetime.format: "YYYY-MM-dd HH:mm:ss"
# 若存在DATE、TIME类型字段需要自定义格式，可按需添加对应配置
# date.format: "YYYY-MM-dd"
# time.format: "HH:mm:ss"

配置添加完成后重启源连接器，DATETIME字段就会直接输出指定格式的字符串，该配置对当前使用的无Schema模式JsonConverter同样生效，不需要调整Converter相关参数。

问题2：Couchbase Sink连接器默认全量覆盖文档，需实现指定嵌套字段增量更新 #

你之前的Sink配置不生效有两个核心原因：一是残留了旧版Couchbase连接器的connection.*前缀配置，会导致配置加载逻辑混乱；二是未指定写入模式，默认走全量覆盖的UPSERT逻辑，无法实现局部字段更新。

解决方案 #

替换原有Sink连接器配置为以下内容，核心新增N1QL子文档更新相关配置：

- name: "our-sink-connector-1"
  config:
    connector.class: "com.couchbase.connect.kafka.CouchbaseSinkConnector"
    tasks.max: "2"
    topics: "our-api.our_db.our_table"
    couchbase.seed.nodes: "dev-couchbase-couchbase-cluster.couchbase.svc.cluster.local"
    couchbase.bootstrap.timeout: "10s"
    couchbase.bucket: "our_bucket"
    couchbase.topic.to.collection: "our-api.our_db.our_table=our_bucket._default.ourCollection"
    couchbase.username: "*******"
    couchbase.password: "*******"
    key.converter: "org.apache.kafka.connect.storage.StringConverter"
    value.converter: "org.apache.kafka.connect.json.JsonConverter"
    value.converter.schemas.enable: "false"
    couchbase.document.id: "${/id}"
    # 以下为子文档更新核心配置
    # 使用N1QL写入模式，支持嵌套字段路径操作
    couchbase.document.mode: "N1QL"
    # 开启字段合并逻辑，不覆盖文档原有未传入的字段
    couchbase.n1ql.merge: "true"
    # 文档不存在时跳过更新，若需要不存在时自动创建可改为true
    couchbase.n1ql.upsert: "false"
    # 字段路径映射：如果Kafka消息中字段名为our_field，需要写入到文档的bill.dateAccepted嵌套路径，就配置这一行
    # 如果消息Value本身就是{"bill": {"dateAccepted": "xxx"}}的结构，可以删掉这行配置
    couchbase.n1ql.field.path: "our_field=bill.dateAccepted"

配置说明 #

• 已移除原有配置中冗余的connection.bucket、connection.cluster_address旧版配置项，新版连接器统一使用couchbase.*前缀配置，避免参数冲突。
• 该配置生效后，Sink端只会更新消息中携带的指定字段，完全不会修改Couchbase中已经预加工完成的其他文档字段，符合局部更新的需求。
• 若后续需要更新多个嵌套字段，直接在couchbase.n1ql.field.path中用逗号分隔多个映射规则即可，例如our_field=bill.dateAccepted,other_field=user.xxx。

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