如何将RDBMS中的数据流式传输至已有的Kafka集群

首先澄清核心认知误区：

Confluent提供的JDBC连接器完全兼容标准Apache Kafka集群，不存在只能对接Confluent自有集群的限制，是完全适配你的场景的最优解之一，不需要自己从零开发同步逻辑。

方案1：Kafka Connect + JDBC源连接器（首推，零业务代码开发） #

• 核心能力：Apache Kafka官方原生组件，自带offset管理、容错重试、并行拉取、数据格式转换能力，完全覆盖你通过查询同步的需求，无需重复造轮子
• 适配你的场景的关键配置：
  1. Kafka Connect集群和你的企业Kafka集群完全解耦，可以独立部署，不需要修改现有Kafka集群的任何配置
  2. 连接器配置中直接指定bootstrap.servers为现有Kafka集群地址，topic参数直接填你已有的目标Topic名称，topic.prefix留空即可
  3. 增量同步可配置incrementing.column.name（自增主键列）或timestamp.column.name（更新时间列）实现增量拉取，全量同步/自定义查询直接写query参数的SQL即可
  4. 提前将Oracle官方JDBC驱动包放到Connect节点的plugin.path目录下即可适配Oracle数据源
• 配置示例：

{
  "name": "oracle-to-kafka-source",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "tasks.max": "1",
    "connection.url": "jdbc:oracle:thin:@//<Oracle地址>:<端口>/<服务名>",
    "connection.user": "<数据库账号>",
    "connection.password": "<数据库密码>",
    "mode": "timestamp",
    "timestamp.column.name": "update_time",
    "query": "SELECT id, col1, col2, update_time FROM 你的业务表",
    "topic.prefix": "",
    "topic": "<你已有的Kafka Topic名称>",
    "bootstrap.servers": "<企业Kafka集群bootstrap地址>",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter"
  }
}

• 维护成本：仅需编写配置文件即可上线，你熟悉Java栈的话也可以很容易开发自定义转换逻辑满足特殊需求

方案2：Flink JDBC Connector + Flink Kafka Sink（适合有复杂数据处理需求的场景） #

• 适用场景：如果你的同步过程需要做数据清洗、字段映射、多表关联等复杂处理，该方案比Kafka Connect灵活性更高，也属于成熟的企业级流处理方案，不需要写底层JDBC和Kafka Producer交互代码
• 实现方式：
  1. 用Flink SQL分别定义JDBC源表映射Oracle业务表，Kafka结果表映射你已有的目标Topic
  2. 直接执行INSERT INTO kafka_sink表 SELECT * FROM jdbc_source表即可完成同步，也可以很方便扩展增量同步、Exactly-Once语义等能力

方案3：Debezium Oracle连接器（可无缝升级为CDC增量同步） #

• 如果你后续需要支持捕获Oracle的全量增删改变更（不需要轮询查询），可以直接使用Debezium的Oracle连接器，同样基于Kafka Connect生态，配置逻辑和JDBC连接器一致，直接对接现有Kafka集群，不需要调整下游Topic的消费逻辑

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