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

2018-05-14 13:51:39本文将演示如何在E-MapReduce上实现将RDS binlog实时同步到Kafka集群中。实验中使用VPC网络环境，以下实例创建时默认都是在VPC环境下。

1. 背景

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

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

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

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

2. 环境准备

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

2.1 准备一个测试RDS数据库

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

2.2 准备一个Kafka集群

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

注意：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组件。

3.2 启动Connector

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

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

这时，我们可以看到一个创建好的connector，如图：

3.3 注意事项

server_id是多少?：你可以在RDS执行”SELECT @@server_id;”查到。

创建connector时可能会出现连接失败，请确保RDS的白名单已经授权了Kafka集群机器访问。

4 测试

4.1 创建一张表

一会之后，Kafka集群中会自动创建一个对应的TOPic

插入几条数据

查看binlog数据

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

结果如图所示：

5. 资料

数据库优化包括哪些相关操作？

此文章主要向大家介绍的是MySQL数据库优化，其中还包括MySQL数据库的性能优化， 常用的SQL语句的优化以及MySQL数据库对INSERT语句进行优化的实际操作方案的描述，望你会有所收获。 MySQL InnoDB 的性能问题讨论 MySQL性能优化 InnoDB delete from xxx速度暴慢原因 推荐圈子: mysql研究 更多相关推荐 1、定期分析表和检查表 分析表的语法如下： 引用 [LOCAL | NO_WRITE_TO_BINLOG] TABLE tb1_name[, tbl_name]... 以上语句用于分析和存储表的关键字分布，分析的结果将可以使得系统得到准确的统计信息，使得SQL能够生成正确的执行计划。 如果用户感觉实际执行计划并不是预期的执行计划，执行一次分析表可能会解决问题。 在分析期间，使用一个读取锁定对表进行锁定。 这对于MyISAM，DBD和InnoDB表有作用。 例如分析一个数据表 引用 table table_name 检查表的语法如下： 引用 TABLE tb1_name[,tbl_name]...[option] = {QUICK | FAST | MEDIUM | EXTENDED | CHANGED} 检查表的作用是检查一个或多个表是否有错误，CHECK TABLE 对MyISAM 和 InnoDB表有作用，对于MyISAM表，关键字统计数据被更新 CHECK TABLE 也可以检查视图是否有错误，比如在视图定义中被引用的表不存在。 2. 定期优化表 MySQL数据库优化表的语法如下： 引用 [LOCAL | NO_WRITE_TO_BINLOG] TABLE tb1_name [,tbl_name]... 如果删除了表的一大部分，或者如果已经对含有可变长度行的表(含有 VARCHAR、BLOB或TEXT列的表)进行更多更改，则应使用OPTIMIZE TABLE命令来进行表优化。 这个命令可以将表中的空间碎片进行合并，并且可以消除由于删除或者更新造成的空间浪费，但OPTIMIZE TABLE 命令只对MyISAM、 BDB 和InnoDB表起作用。 例如： optimize table table_name 注意： analyze、check、optimize执行期间将对表进行锁定，因此一定注意要在数据库不繁忙的时候执行相关的操作。 常用的SQL优化 我们在开发的时候常常用到的SQL语句，无非是INSERT、GROUPBY等等。 对于这些SQL语句，我们怎么进行优化？ 1. 大批量插入数据 当用load命令导入数据的时候，适当的设置可以提高导入的速度。 对于MyISAM存储引擎的表，可以通过如下方式快速的导入大量的数据 引用 TABLE tb1_name DISABLE KEYS; the data TABLE tb1_name ENABLE KEYS; DISABLE KEYS 和 ENABLE KEYS 用来打开或者关闭MyISAM表非唯一索引的更新。 在导入大量的数据到一个非空的MyISAM表时，通过设置这两个命令，可以提高导入的效率。 对于导入大量的数据到一个空的MyISAM表时，默认就是先导入数据然后才创建索引的，索引不用进行设置。 引用 data infile /home/mysql/text_txt into table text 对于InnoDB类型的表，这种方式不能提高导入数据的效率，但也有几种针对InnoDB类型的表进行MySQL数据库优化的方式。 1. 因为InnoDB类型的表式按照主键的顺序保存的，所以将导入的数据按照主键的顺序排序，可以有效提高导入数据的效率。 2. 在导入数据前执行 SET UNIQUE_CHECKS=0，关闭唯一性校验，在导入结束后执行SET UNIQUE_CHECKS=1，恢复唯一性校验，可以提高导入的效率。 3. 如果应用使用自动提交的方式，建议在导入前执行SET AUTOCOMMIT=0，关闭自动提交，导入结束后执行SET AUTOCOMMIT=1，打开自动提交，也可以提高导入效率。 MySQL数据库优化INSERT语句 当进行数据INSERT的时候，可以考虑采用以下几种方式进行优化 1. 如果同时从一个客户插入很多行，尽量使用多个值表的INSERT语句，这种方式将大大缩短客户端与数据库的链接、关闭等消耗，使得效率比分开执行的单个INSERT语句快. 例如： into test values(1,2) into test values(3,4) into test values(5,6) 将上面三句改为:insert into test values(1,2),(3,4),(5,6)...... 2. 如果从不同客户插入很多行，能通过使用INSERT DELAYED 语句得到更高的速度。 DELAYED 的含义是让INSERT 语句马上执行，其实数据都被放在内存的队列中，并没有真正写入磁盘，这比每条语句分别插入要快得多；LOW_PRIORITY刚好相反，在所有其他用户对表的读写完后才进行插入。 3. 将索引文件和数据文件分在不同的磁盘上存放 4. 如果进行批量插入，可以增加bulk_insert_buffer_size变量值的方法来提高速度，但是，这只能对于MyISAM表使用。 5. 当从一个文本文件中装载一个表时，使用LOAD DATA INFILE。 这通常比使用很多insert语句快20倍左右。 以上的相关内容就是对MySQL数据库优化方法的介绍，望你能有所收获。

如何再application.xml文件中配置访问mysql主从分离

1、主数据库服务器配置进入主数据库服务器安装目录,打开,在文件末尾增加如下配置：#数据库ID号， 为1时表示为Master,其中master_id必须为1到232–1之间的一个正整数值; server-id = 1#启用二进制日志；log-bin=mysql-bin#需要同步的二进制数据库名；binlog-do-db=minishop#不同步的二进制数据库名,如果不设置可以将其注释掉;binlog-ignore-db=information_schemabinlog-ignore-db=mysqlbinlog-ignore-db=personalsite

mysql数据库如何备份?

目前 MySQL 支持的免费备份工具有：mysqldump、mysqlhotcopy，还可以用 SQL 语法进行备份：BACKUP TABLE 或者 SELECT INTO OUTFILE，又或者备份二进制日志（binlog），还可以是直接拷贝数据文件和相关的配置文件。 MyISAM 表是保存成文件的形式，因此相对比较容易备份，上面提到的几种方法都可以使用。 Innodb 所有的表都保存在同一个数据文件 ibdata1 中（也可能是多个文件，或者是独立的表空间文件），相对来说比较不好备份，免费的方案可以是拷贝数据文件、备份 binlog，或者用 mysqldump。