如何使用Kafka Connect实现同步RDS binlog数据

本文涉及的产品
EMR Serverless StarRocks,5000CU*H 48000GB*H
简介: 本文介绍如何在E-MapReduce上使用Kafka Connect实现同步RDS binlog数据

1. 背景

在我们的业务开发中,往往会碰到下面这个场景:

  • 业务更新数据写到数据库中
  • 业务更新数据需要实时传递给下游依赖处理

所以传统的处理架构可能会这样:

image

但这个架构也存在着不少弊端:我们需要在项目中维护很多发送消息的代码。新增或者更新消息都会带来不少维护成本。所以,更好的处理方式应该是直接将数据库的数据接入到流式系统中,如下图:
image

本文将演示如何在E-MapReduce上实现将RDS binlog实时同步到Kafka集群中。

2. 环境准备

实验中使用VPC网络环境,以下实例创建时默认都是在VPC环境下。

2.1 准备一个测试RDS数据库

创建一个RDS实例,版本选择5.7。这里不赘述如何创建RDS,详细流程请参考RDS文档。创建完如图:
image

2.2 准备一个Kafka集群

创建一个E-MapReduce Kafka集群,版本选择EMR-3.11.0。需要注意,这里必须选择EMR-3.11.0以上版本,否则不会默认安装启动Kafka Connect服务。详细创建流程请参考E-MapReduce文档。创建完如图:
image

注意:RDS实例和E-MapReduce Kafka集群最好在同一个VPC中,否则需要打通两个VPC之间的网络。

3. Kafka Connect

3.1 Connector

Kafka Connect是一个用于Kafka和其他数据系统之间进行数据传输的工具,它可以实现基于Kafka的数据管道,打通上下游数据源。我们需要做的就是在Kafka Connect服务上运行一个Connector,这个Connector是具体实现如何从/向数据源中读/写数据。Confluent提供了很多Connector实现,你可以在这里下载。不过今天我们使用Debezium提供的一个MySQL Connector插件,下载地址

  • 下载这个插件,并将解压出来的jar包全部拷贝到kafka lib目录下。注意:需要将这些jar包拷贝到Kafka集群所有机器上。
  • 在Kafka集群的服务列表中重启Kafka Connect组件。
    image

3.2 启动Connector

在创建connector前,我们需要做一番配置,这里罗列一些Debezium MySQL Connector的主要配置项:

database.hostname=x.x.x.x
database.port=3306
database.user=tom
database.password=password
database.server.id=123456
database.server.name=fullfillment
database.whitelist=inventory
database.history.kafka.bootstrap.servers=y.y.y.y:9092
database.history.kafka.topic=dbhistory.fullfillment
include.schema.changes=true

登录到Kafka集群,配置并创建一个connector,命令如下:

curl -X POST -H "Content-Type: application/json" 
--data '{"name": "rds-binlog", 
"config": {"connector.class":"io.debezium.connector.mysql.MySqlConnector", 
"database.hostname": "x.x.x.x", "database.port": "3306", 
"database.user": "tom", "database.password": "password", 
"database.server.id": "123456", "database.server.name": "fulfillment", 
"database.history.kafka.bootstrap.servers": "y.y.y.y:9092", 
"database.history.kafka.topic": "dbhistory.fullfillment", 
"include.schema.changes": "true"}}' 
http://emr-worker-1:8083/connectors

这时,我们可以看到一个创建好的connector,如图:
image

3.3 注意事项

  • server_id是多少?:你可以在RDS执行"SELECT @@server_id;"查到。
  • 创建connector时可能会出现连接失败,请确保RDS的白名单已经授权了Kafka集群机器访问。

4 测试

4.1 创建一张表

image

一会之后,Kafka集群中会自动创建一个对应的topic
image

插入几条数据

image

查看binlog数据

查看fulfillment.mugen.students这个topic,是否有刚刚新插入的数据

kafka-console-consumer.sh --zookeeper emr-header-1:2181/kafka-1.0.1 
--topic fulfillment.mugen.students --from-beginning

结果如图所示:

image

5. 资料

目录
相关文章
|
2月前
|
安全 关系型数据库 MySQL
如何将数据从MySQL同步到其他系统
【10月更文挑战第17天】如何将数据从MySQL同步到其他系统
209 0
|
2月前
|
消息中间件 关系型数据库 MySQL
大数据-117 - Flink DataStream Sink 案例:写出到MySQL、写出到Kafka
大数据-117 - Flink DataStream Sink 案例:写出到MySQL、写出到Kafka
171 0
|
2月前
|
SQL 存储 关系型数据库
Mysql主从同步 清理二进制日志的技巧
Mysql主从同步 清理二进制日志的技巧
31 1
|
3月前
|
消息中间件 canal 关系型数据库
Maxwell:binlog 解析器,轻松同步 MySQL 数据
Maxwell:binlog 解析器,轻松同步 MySQL 数据
361 11
|
4月前
|
关系型数据库 MySQL Linux
mysql 主从同步 实现增量备份
【8月更文挑战第28天】mysql 主从同步 实现增量备份
52 3
|
4月前
|
消息中间件 关系型数据库 MySQL
实时计算 Flink版产品使用问题之使用CTAS同步MySQL到Hologres时出现的时区差异,该如何解决
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
4月前
|
SQL 存储 关系型数据库
实时计算 Flink版产品使用问题之同步MySQL多张表的过程中,内存释放依赖于什么
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
2月前
|
消息中间件 存储 运维
为什么说Kafka还不是完美的实时数据通道
【10月更文挑战第19天】Kafka 虽然作为数据通道被广泛应用,但在实时性、数据一致性、性能及管理方面存在局限。数据延迟受消息堆积和分区再平衡影响;数据一致性难以达到恰好一次;性能瓶颈在于网络和磁盘I/O;管理复杂性涉及集群配置与版本升级。
|
2月前
|
消息中间件 Java Kafka
Flink-04 Flink Java 3分钟上手 FlinkKafkaConsumer消费Kafka数据 进行计算SingleOutputStreamOperatorDataStreamSource
Flink-04 Flink Java 3分钟上手 FlinkKafkaConsumer消费Kafka数据 进行计算SingleOutputStreamOperatorDataStreamSource
51 1
|
4月前
|
消息中间件 Java Kafka
Kafka不重复消费的终极秘籍!解锁幂等性、偏移量、去重神器,让你的数据流稳如老狗,告别数据混乱时代!
【8月更文挑战第24天】Apache Kafka作为一款领先的分布式流处理平台,凭借其卓越的高吞吐量与低延迟特性,在大数据处理领域中占据重要地位。然而,在利用Kafka进行数据处理时,如何有效避免重复消费成为众多开发者关注的焦点。本文深入探讨了Kafka中可能出现重复消费的原因,并提出了四种实用的解决方案:利用消息偏移量手动控制消费进度;启用幂等性生产者确保消息不被重复发送;在消费者端实施去重机制;以及借助Kafka的事务支持实现精确的一次性处理。通过这些方法,开发者可根据不同的应用场景灵活选择最适合的策略,从而保障数据处理的准确性和一致性。
313 9