RocketMQ Connect实战1

MySQL Source(CDC) - >RocketMQ Connect -> MySQL Sink(JDBC)

准备

启动RocketMQ

  1. Linux/Unix/Mac
  2. 64bit JDK 1.8+;
  3. Maven 3.2.x或以上版本;
  4. 启动 RocketMQ;

tips : ${ROCKETMQ_HOME} 位置说明

bin-release.zip 版本：/rocketmq-all-4.9.4-bin-release

source-release.zip 版本：/rocketmq-all-4.9.4-source-release/distribution

启动Connect

Connector插件编译

Debezium RocketMQ Connector

  1. $ cd rocketmq-connect/connectors/rocketmq-connect-debezium/
  2. $ mvn clean package -Dmaven.test.skip=true

将 Debezium MySQL RocketMQ Connector 编译好的包放入Runtime加载目录。命令如下：

  1. mkdir -p /usr/local/connector-plugins
  2. cp rocketmq-connect-debezium-mysql/target/rocketmq-connect-debezium-mysql-0.0.1-SNAPSHOT-jar-with-dependencies.jar /usr/local/connector-plugins

JDBC Connector

将 JDBC Connector 编译好的包放入Runtime加载目录。命令如下：

  1. $ cd rocketmq-connect/connectors/rocketmq-connect-jdbc/
  2. $ mvn clean package -Dmaven.test.skip=true
  3. cp rocketmq-connect-jdbc/target/rocketmq-connect-jdbc-0.0.1-SNAPSHOT-jar-with-dependencies.jar /usr/local/connector-plugins

启动Connect Runtime

  1. cd  rocketmq-connect
  3. mvn -Prelease-connect -DskipTests clean install -U

修改配置connect-standalone.conf ，重点配置如下

  1. $ cd distribution/target/rocketmq-connect-0.0.1-SNAPSHOT/rocketmq-connect-0.0.1-SNAPSHOT
  2. $ vim conf/connect-standalone.conf
  1. workerId=standalone-worker
  2. storePathRootDir=/tmp/storeRoot
  4. ## Http port for user to access REST API
  5. httpPort=8082
  7. # Rocketmq namesrvAddr
  8. namesrvAddr=localhost:9876
  10. # RocketMQ acl
  11. aclEnable=false
  12. accessKey=rocketmq
  13. secretKey=12345678
  15. autoCreateGroupEnable=false
  16. clusterName="DefaultCluster"
  18. # 核心配置，将之前编译好debezium包的插件目录配置在此；
  19. # Source or sink connector jar file dir,The default value is rocketmq-connect-sample
  20. pluginPaths=/usr/local/connector-plugins
  1. cd distribution/target/rocketmq-connect-0.0.1-SNAPSHOT/rocketmq-connect-0.0.1-SNAPSHOT
  3. sh bin/connect-standalone.sh -c conf/connect-standalone.conf &

MySQL镜像

使用debezium的MySQL docker搭建环境MySQL数据库

  1. docker run -it --rm --name mysql -p 3306:3306 -e MYSQL_ROOT_PASSWORD=debezium -e MYSQL_USER=mysqluser -e MYSQL_PASSWORD=mysqlpw quay.io/debezium/example-mysql:1.9

MySQL信息

端口：3306

账号：root/debezium

slave:debezium/dbz

测试数据

通过root/debezium账号登录数据库

源数据库表：inventory.employee

  1. CREATE database inventory;
  3. use inventory;
  4. CREATE TABLE `employee` (
  5. `id` bigint NOT NULL AUTO_INCREMENT,
  6. `name` varchar(128) DEFAULT NULL,
  7. `howold` int DEFAULT NULL,
  8. `male` int DEFAULT NULL,
  9. `company` varchar(128) DEFAULT NULL,
  10. `money` double DEFAULT NULL,
  11. `begin_time` datetime DEFAULT NULL,
  12. `modify_time` timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '修改时间',
  13. `decimal_test` decimal(11,2) DEFAULT NULL COMMENT 'test decimal type',
  14. PRIMARY KEY (`id`)
  15. ) ENGINE=InnoDB AUTO_INCREMENT=16 DEFAULT CHARSET=utf8;
  19. INSERT INTO `employee` VALUES (1, 'name-01', 24, 6, 'company', 9987, '2021-12-22 08:00:00', '2022-06-14 18:20:11', 321.11);
  20. INSERT INTO `employee` VALUES (2, 'name-02', 19, 7, 'company', 32232, '2021-12-29 08:00:00', '2022-06-14 18:18:47', 77.12);
  21. INSERT INTO `employee` VALUES (8, 'name-03', 20, 1, NULL, 0, NULL, '2022-06-14 18:26:05', 11111.00);
  22. INSERT INTO `employee` VALUES (9, 'name-04', 21, 1, 'company', 12345, '2021-12-24 20:44:10', '2022-06-14 18:20:02', 123.12);
  23. INSERT INTO `employee` VALUES (11, 'name-05', 50, 2, 'company', 33333, '2021-12-24 22:14:52', '2022-06-14 18:19:58', 123.12);
  24. INSERT INTO `employee` VALUES (12, 'name-06', 19, 3, NULL, 0, NULL, '2022-06-14 18:26:12', 111233.00);
  25. INSERT INTO `employee` VALUES (13, 'name-07', 20, 4, 'company', 3237, '2021-12-29 01:31:03', '2022-06-14 18:19:27', 52.00);
  26. INSERT INTO `employee` VALUES (14, 'name-08', 25, 15, 'company', 32255, '2022-02-08 19:06:39', '2022-06-14 18:18:32', 0.00);
  27. INSERT INTO `employee` VALUES (15, NULL, 0, 0, NULL, 0, NULL, '2022-06-14 20:13:29', NULL);

目标库：inventory_2.employee

  1. CREATE database inventory_2;
  2. use inventory_2;
  3. CREATE TABLE `employee` (
  4. `id` bigint NOT NULL AUTO_INCREMENT,
  5. `name` varchar(128) DEFAULT NULL,
  6. `howold` int DEFAULT NULL,
  7. `male` int DEFAULT NULL,
  8. `company` varchar(128) DEFAULT NULL,
  9. `money` double DEFAULT NULL,
  10. `begin_time` datetime DEFAULT NULL,
  11. `modify_time` timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '修改时间',
  12. `decimal_test` decimal(11,2) DEFAULT NULL COMMENT 'test decimal type',
  13. PRIMARY KEY (`id`)
  14. ) ENGINE=InnoDB AUTO_INCREMENT=16 DEFAULT CHARSET=utf8;

启动Connector

启动Debezium source connector

同步原表数据：inventory.employee 作用：通过解析MySQL binlog 封装成通用的ConnectRecord对象，发送的RocketMQ Topic当中

  1. curl -X POST -H "Content-Type: application/json" http://127.0.0.1:8082/connectors/MySQLCDCSource -d '{
  2. "connector.class": "org.apache.rocketmq.connect.debezium.mysql.DebeziumMysqlConnector",
  3. "max.task": "1",
  4. "connect.topicname": "debezium-mysql-source-topic",
  5. "kafka.transforms": "Unwrap",
  6. "kafka.transforms.Unwrap.delete.handling.mode": "none",
  7. "kafka.transforms.Unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
  8. "kafka.transforms.Unwrap.add.headers": "op,source.db,source.table",
  9. "database.history.skip.unparseable.ddl": true,
  10. "database.history.name.srv.addr": "localhost:9876",
  11. "database.history.rocketmq.topic": "db-history-debezium-topic",
  12. "database.history.store.only.monitored.tables.ddl": true,
  13. "include.schema.changes": false,
  14. "database.server.name": "dbserver1",
  15. "database.port": 3306,
  16. "database.hostname": "数据库ip",
  17. "database.connectionTimeZone": "UTC",
  18. "database.user": "debezium",
  19. "database.password": "dbz",
  20. "table.include.list": "inventory.employee",
  21. "max.batch.size": 50,
  22. "database.include.list": "inventory",
  23. "snapshot.mode": "when_needed",
  24. "database.server.id": "184054",
  25. "key.converter": "org.apache.rocketmq.connect.runtime.converter.record.json.JsonConverter",
  26. "value.converter": "org.apache.rocketmq.connect.runtime.converter.record.json.JsonConverter"
  27. }'

启动 jdbc sink connector

作用：通过消费Topic中的数据，通过JDBC协议写入到目标表当中

  1. curl -X POST -H "Content-Type: application/json" http://127.0.0.1:8082/connectors/jdbcmysqlsinktest -d '{
  2.   "connector.class": "org.apache.rocketmq.connect.jdbc.connector.JdbcSinkConnector",
  3.   "max.task": "2",
  4.   "connect.topicnames": "debezium-mysql-source",
  5.   "connection.url": "jdbc:mysql://数据库ip:3306/inventory_2",
  6.   "connection.user": "root",
  7.   "connection.password": "debezium",
  8.   "pk.fields": "id",
  9.   "table.name.from.header": "true",
  10.   "pk.mode": "record_key",
  11.   "insert.mode": "UPSERT",
  12.   "db.timezone": "UTC",
  13.   "table.types": "TABLE",
  14.   "errors.deadletterqueue.topic.name": "dlq-topic",
  15.   "errors.log.enable": "true",
  16.   "errors.tolerance": "ALL",
  17.   "delete.enabled": "true",
  18.   "key.converter": "org.apache.rocketmq.connect.runtime.converter.record.json.JsonConverter",
  19.   "value.converter": "org.apache.rocketmq.connect.runtime.converter.record.json.JsonConverter"
  20. }'

以上两个Connector任务创建成功以后 通过root/debezium账号登录数据库

对源数据库表：inventory.employee增删改 即可同步到目标办inventory_2.employee