Binlog Load
Binlog Load提供了一种使Doris增量同步用户在Mysql数据库的对数据更新操作的CDC(Change Data Capture)功能。
适用场景
INSERT/UPDATE/DELETE支持
过滤Query
暂不兼容DDL语句
名词解释
Frontend（FE）：Doris 系统的元数据和调度节点。在导入流程中主要负责导入 plan 生成和导入任务的调度工作。
Backend（BE）：Doris 系统的计算和存储节点。在导入流程中主要负责数据的 ETL 和存储。
Canal：阿里巴巴开源的Mysql Binlog日志解析工具。提供增量数据订阅&消费等功能。
Batch：Canal发送到客户端的一批数据，具有全局唯一自增的ID。
SyncJob：用户提交的一个数据同步作业。
Receiver: 负责订阅并接收Canal的数据。
Consumer: 负责分发Receiver接收的数据到各个Channel。
Channel: 接收Consumer分发的数据的渠道，创建发送数据的子任务，控制单个表事务的开启、提交、终止。
Task：Channel向Be发送数据的子任务。
基本原理
在第一期的设计中，Binlog Load需要依赖canal作为中间媒介，让canal伪造成一个从节点去获取Mysql主节点上的Binlog并解析，再由Doris去获取Canal上解析好的数据，主要涉及Mysql端、Canal端以及Doris端，总体数据流向如下：
+---------------------------------------------+
| Mysql |
+----------------------+----------------------+
| Binlog
+----------------------v----------------------+
| Canal Server |
+-------------------+-----^-------------------+
Get | | Ack
+-------------------|-----|-------------------+
| FE | | |
| +-----------------|-----|----------------+ |
| | Sync Job | | | |
| | +------------v-----+-----------+ | |
| | | Canal Client | | |
| | | +-----------------------+ | | |
| | | | Receiver | | | |
| | | +-----------------------+ | | |
| | | +-----------------------+ | | |
| | | | Consumer | | | |
| | | +-----------------------+ | | |
| | +------------------------------+ | |
| +----+---------------+--------------+----+ |
| | | | |
| +----v-----+ +-----v----+ +-----v----+ |
| | Channel1 | | Channel2 | | Channel3 | |
| | [Table1] | | [Table2] | | [Table3] | |
| +----+-----+ +-----+----+ +-----+----+ |
| | | | |
| +--|-------+ +---|------+ +---|------+|
| +---v------+| +----v-----+| +----v-----+||
| +----------+|+ +----------+|+ +----------+|+|
| | Task |+ | Task |+ | Task |+ |
| +----------+ +----------+ +----------+ |
+----------------------+----------------------+
| | |
+----v-----------------v------------------v---+
| Coordinator |
| BE |
+----+-----------------+------------------+---+
| | |
+----v---+ +---v----+ +----v---+
| BE | | BE | | BE |
+--------+ +--------+ +--------+
如上图，用户向FE提交一个数据同步作业。
FE会为每个数据同步作业启动一个canal client，来向canal server端订阅并获取数据。
client中的receiver将负责通过Get命令接收数据，每获取到一个数据batch，都会由consumer根据对应表分发到不同的channel，每个channel都会为此数据batch产生一个发送数据的子任务Task。
在FE上，一个Task是channel向BE发送数据的子任务，里面包含分发到当前channel的同一个batch的数据。
channel控制着单个表事务的开始、提交、终止。一个事务周期内，一般会从consumer获取到多个batch的数据，因此会产生多个向BE发送数据的子任务Task，在提交事务成功前，这些Task不会实际生效。
满足一定条件时（比如超过一定时间、达到提交最大数据大小），consumer将会阻塞并通知各个channel提交事务。
当且仅当所有channel都提交成功，才会通过Ack命令通知canal并继续获取并消费数据。
如果有任意channel提交失败，将会重新从上一次消费成功的位置获取数据并再次提交（已提交成功的channel不会再次提交以保证幂等性）。
整个数据同步作业中，FE通过以上流程不断的从canal获取数据并提交到BE，来完成数据同步。
配置Mysql端
在Mysql Cluster模式的主从同步中，二进制日志文件(Binlog)记录了主节点上的所有数据变化，数据在Cluster的多个节点间同步、备份都要通过Binlog日志进行，从而提高集群的可用性。架构通常由一个主节点(负责写)和一个或多个从节点(负责读)构成，所有在主节点上发生的数据变更将会复制给从节点。
注意：目前必须要使用Mysql 5.7及以上的版本才能支持Binlog Load功能。
要打开mysql的二进制binlog日志功能，则需要编辑my.cnf配置文件设置一下。
[mysqld]
log-bin = mysql-bin # 开启 binlog
binlog-format=ROW # 选择 ROW 模式
Mysql端说明
在Mysql上，Binlog命名格式为mysql-bin.000001、mysql-bin.000002… ，满足一定条件时mysql会去自动切分Binlog日志：
mysql重启了
客户端输入命令flush logs
binlog文件大小超过1G
要定位Binlog的最新的消费位置，可以通过binlog文件名和position(偏移量)。
例如，各个从节点上会保存当前消费到的binlog位置，方便随时断开连接、重新连接和继续消费。
--------------------- ---------------------
| Slave | read | Master |
| FileName/Position | <<<--------------------------- | Binlog Files |
--------------------- ---------------------
对于主节点来说，它只负责写入Binlog，多个从节点可以同时连接到一台主节点上，消费Binlog日志的不同部分，互相之间不会影响。
Binlog日志支持两种主要格式(此外还有混合模式mixed-based):
statement-based格式: Binlog只保存主节点上执行的sql语句，从节点将其复制到本地重新执行
row-based格式: Binlog会记录主节点的每一行所有列的数据的变更信息，从节点会复制并执行每一行的变更到本地
第一种格式只写入了执行的sql语句，虽然日志量会很少，但是有下列缺点
没有保存每一行实际的数据
在主节点上执行的UDF、随机、时间函数会在从节点上结果不一致
limit语句执行顺序可能不一致
因此我们需要选择第二种格式，才能从Binlog日志中解析出每一行数据。
在row-based格式下，Binlog会记录每一条binlog event的时间戳，server id，偏移量等信息，如下面一条带有两条insert语句的事务:
begin;
insert into canal_test.test_tbl values (3, 300);
insert into canal_test.test_tbl values (4, 400);
commit;
对应将会有四条binlog event，其中一条begin event，两条insert event，一条
commit event：
SET TIMESTAMP=1538238301/*!*/;
BEGIN
/*!*/.
# at 211935643
# at 211935698
#180930 0:25:01 server id 1 end_log_pos 211935698 Table_map: 'canal_test'.'test_tbl' mapped to number 25
#180930 0:25:01 server id 1 end_log_pos 211935744 Write_rows: table-id 25 flags: STMT_END_F
...
'/*!*/;
### INSERT INTO canal_test.test_tbl
### SET
### @1=1
### @2=100
# at 211935744
#180930 0:25:01 server id 1 end_log_pos 211935771 Xid = 2681726641
...
'/*!*/;
### INSERT INTO canal_test.test_tbl
### SET
### @1=2
### @2=200
# at 211935771
#180930 0:25:01 server id 1 end_log_pos 211939510 Xid = 2681726641
COMMIT/*!*/;
如上图所示，每条Insert event中包含了修改的数据。在进行Delete/Update操作时，一条event还能包含多行数据，使得Binlog日志更加的紧密。
开启GTID模式 [可选]
一个全局事务Id(global transaction identifier)标识出了一个曾在主节点上提交过的事务，在全局都是唯一有效的。开启了Binlog后，GTID会被写入到Binlog文件中，与事务一一对应。
要打开mysql的GTID模式，则需要编辑my.cnf配置文件设置一下
gtid-mode=on // 开启gtid模式
enforce-gtid-consistency=1 // 强制gtid和事务的一致性
在GTID模式下，主服务器可以不需要Binlog的文件名和偏移量，就能很方便的追踪事务、恢复数据、复制副本。
在GTID模式下，由于GTID的全局有效性，从节点将不再需要通过保存文件名和偏移量来定位主节点上的Binlog位置，而通过数据本身就可以定位了。在进行数据同步中，从节点会跳过执行任意被识别为已执行的GTID事务。
GTID的表现形式为一对坐标, source_id标识出主节点，transaction_id表示此事务在主节点上执行的顺序(最大263-1)。
GTID = source_id:transaction_id
例如，在同一主节点上执行的第23个事务的gtid为
3E11FA47-71CA-11E1-9E33-C80AA9429562:23
配置Canal端
canal是属于阿里巴巴otter项目下的一个子项目，主要用途是基于 MySQL 数据库增量日志解析，提供增量数据订阅和消费，用于解决跨机房同步的业务场景，建议使用canal 1.1.5及以上版本，下载地址 (opens new window)，下载完成后，请按以下步骤完成部署。
解压canal deployer
在conf文件夹下新建目录并重命名，作为instance的根目录，目录名即后文提到的destination
修改instance配置文件（可拷贝conf/example/instance.properties）
vim conf/{your destination}/instance.properties
## canal instance serverId
canal.instance.mysql.slaveId = 1234
## mysql adress
canal.instance.master.address = 127.0.0.1:3306
## mysql username/password
canal.instance.dbUsername = canal
canal.instance.dbPassword = canal
启动
sh bin/startup.sh
验证启动成功
cat logs/{your destination}/{your destination}.log
2013-02-05 22:50:45.636 [main] INFO c.a.o.c.i.spring.support.PropertyPlaceholderConfigurer - Loading properties file from class path resource [canal.properties]
2013-02-05 22:50:45.641 [main] INFO c.a.o.c.i.spring.support.PropertyPlaceholderConfigurer - Loading properties file from class path resource [xxx/instance.properties]
2013-02-05 22:50:45.803 [main] INFO c.a.otter.canal.instance.spring.CanalInstanceWithSpring - start CannalInstance for 1-xxx
2013-02-05 22:50:45.810 [main] INFO c.a.otter.canal.instance.spring.CanalInstanceWithSpring - start successful....
canal端说明
canal通过伪造自己的mysql dump协议，去伪装成一个从节点，获取主节点的Binlog日志并解析。
canal server上可启动多个instance，一个instance可看作一个从节点，每个instance由下面几个部分组成：
-------------------------------------------------
| Server |
| -------------------------------------------- |
| | Instance 1 | |
| | ----------- ----------- ----------- | |
| | | Parser | | Sink | | Store | | |
| | ----------- ----------- ----------- | |
| | ----------------------------------- | |
| | | MetaManager | | |
| | ----------------------------------- | |
| -------------------------------------------- |
-------------------------------------------------
parser：数据源接入，模拟slave协议和master进行交互，协议解析
sink：parser和store链接器，进行数据过滤，加工，分发的工作
store：数据存储
meta manager：元数据管理模块
每个instance都有自己在cluster内的唯一标识，即server Id。
在canal server内，instance用字符串表示，此唯一字符串被记为destination，canal client需要通过destination连接到对应的instance。
注意：canal client和canal instance是一一对应的，Binlog Load已限制多个数据同步作业不能连接到同一个destination。
数据在instance内的流向是binlog -> parser -> sink -> store。
instance通过parser模块解析binlog日志，解析出来的数据缓存在store里面，当用户向FE提交一个数据同步作业时，会启动一个canal client订阅并获取对应instance中的store内的数据。
store实际上是一个环形的队列，用户可以自行配置它的长度和存储空间。
store
store通过三个指针去管理队列内的数据：
get指针：get指针代表客户端最后获取到的位置。
ack指针：ack指针记录着最后消费成功的位置。
put指针：代表sink模块最后写入store成功的位置。
canal client异步获取store中数据
get 0 get 1 get 2 put
| | | ...... |
v v v v
--------------------------------------------------------------------- store环形队列
^ ^
| |
ack 0 ack 1
canal client调用get命令时，canal server会产生数据batch发送给client，并右移get指针，client可以获取多个batch，直到get指针赶上put指针为止。
当消费数据成功时，client会返回ack + batch Id通知已消费成功了，并右移ack指针，store会从队列中删除此batch的数据，腾出空间来从上游sink模块获取数据，并右移put指针。
当数据消费失败时，client会返回rollback通知消费失败，store会将get指针重置左移到ack指针位置，使下一次client获取的数据能再次从ack指针处开始。
和Mysql中的从节点一样，canal也需要去保存client最新消费到的位置。canal中所有元数据(如GTID、Binlog位置)都是由MetaManager去管理的，目前元数据默认以json格式持久化在instance根目录下的meta.dat文件内。
基本操作
#配置目标表属性
用户需要先在Doris端创建好与Mysql端对应的目标表
Binlog Load只能支持Unique类型的目标表，且必须激活目标表的Batch Delete功能。
开启Batch Delete的方法可以参考help alter table中的批量删除功能。
示例：
-- create target table
CREATE TABLE `test1` (
`a` int(11) NOT NULL COMMENT "",
`b` int(11) NOT NULL COMMENT ""
) ENGINE=OLAP
UNIQUE KEY(`a`)
COMMENT "OLAP"
DISTRIBUTED BY HASH(`a`) BUCKETS 8;
-- enable batch delete
ALTER TABLE canal_test.test1 ENABLE FEATURE "BATCH_DELETE";
创建同步作业
创建数据同步作业的的详细语法可以连接到 Doris 后，执行 HELP CREATE SYNC JOB; 查看语法帮助。这里主要详细介绍，创建作业时的注意事项。
job_name
job_name是数据同步作业在当前数据库内的唯一标识，相同job_name的作业只能有一个在运行。
channel_desc
channel_desc用来定义任务下的数据通道，可表示mysql源表到doris目标表的映射关系。在设置此项时，如果存在多个映射关系，必须满足mysql源表应该与doris目标表是一一对应关系，其他的任何映射关系（如一对多关系），检查语法时都被视为不合法。
column_mapping
column_mapping主要指mysql源表和doris目标表的列之间的映射关系，如果不指定，FE会默认源表和目标表的列按顺序一一对应。但是我们依然建议显式的指定列的映射关系，这样当目标表的结构发生变化（比如增加一个 nullable 的列），数据同步作业依然可以进行。否则，当发生上述变动后，因为列映射关系不再一一对应，导入将报错。
binlog_desc
binlog_desc中的属性定义了对接远端Binlog地址的一些必要信息，目前可支持的对接类型只有canal方式，所有的配置项前都需要加上canal前缀。
canal.server.ip: canal server的地址
canal.server.port: canal server的端口
canal.destination: 前文提到的instance的字符串标识
canal.batchSize: 每批从canal server处获取的batch大小的最大值，默认8192
canal.username: instance的用户名
canal.password: instance的密码
canal.debug: 设置为true时，会将batch和每一行数据的详细信息都打印出来，会影响性能。
#查看作业状态
查看作业状态的具体命令和示例可以通过 HELP SHOW SYNC JOB; 命令查看。
返回结果集的参数意义如下：
State
作业当前所处的阶段。作业状态之间的转换如下图所示：
+-------------+
create job | PENDING | resume job
+-----------+ <-------------+
| +-------------+ |
+----v-------+ +-------+----+
| RUNNING | pause job | PAUSED |
| +-----------------------> |
+----+-------+ run error +-------+----+
| +-------------+ |
| | CANCELLED | |
+-----------> <-------------+
stop job +-------------+ stop job
system error
作业提交之后状态为PENDING，由FE调度执行启动canal client后状态变成RUNNING，用户可以通过 STOP/PAUSE/RESUME 三个命令来控制作业的停止，暂停和恢复，操作后作业状态分别为CANCELLED/PAUSED/RUNNING。
作业的最终阶段只有一个CANCELLED，当作业状态变为CANCELLED后，将无法再次恢复。当作业发生了错误时，若错误是不可恢复的，状态会变成CANCELLED，否则会变成PAUSED。
Channel
作业所有源表到目标表的映射关系。
Status
当前binlog的消费位置(若设置了GTID模式，会显示GTID)，以及doris端执行时间相比mysql端的延迟时间。
JobConfig
对接的远端服务器信息，如canal server的地址与连接instance的destination
控制作业
用户可以通过 STOP/PAUSE/RESUME 三个命令来控制作业的停止，暂停和恢复。可以通过HELP STOP SYNC JOB; HELP PAUSE SYNC JOB; 以及 HELP RESUME SYNC JOB; 三个命令查看帮助和示例。
相关参数
CANAL配置
下面配置属于canal端的配置，主要通过修改 conf 目录下的 canal.properties 调整配置值。
canal.ip
canal server的ip地址
canal.port
canal server的端口
canal.instance.memory.buffer.size
canal端的store环形队列的队列长度，必须设为2的幂次方，默认长度16384。此值等于canal端能缓存event数量的最大值，也直接决定了Doris端一个事务内所能容纳的最大event数量。建议将它改的足够大，防止Doris端一个事务内能容纳的数据量上限太小，导致提交事务太过频繁造成数据的版本堆积。
canal.instance.memory.buffer.memunit
canal端默认一个event所占的空间，默认空间为1024 bytes。此值乘上store环形队列的队列长度等于store的空间最大值，比如store队列长度为16384，则store的空间为16MB。但是，一个event的实际大小并不等于此值，而是由这个event内有多少行数据和每行数据的长度决定的，比如一张只有两列的表的insert event只有30字节，但delete event可能达到数千字节，这是因为通常delete event的行数比insert event多。
FE配置
下面配置属于数据同步作业的系统级别配置，主要通过修改 fe.conf 来调整配置值。
enable_create_sync_job
开启数据同步作业功能。默认为 false，关闭此功能。
sync_commit_interval_second
提交事务的最大时间间隔。若超过了这个时间channel中还有数据没有提交，consumer会通知channel提交事务。
min_sync_commit_size
提交事务需满足的最小event数量。若Fe接收到的event数量小于它，会继续等待下一批数据直到时间超过了sync_commit_interval_second为止。默认值是10000个events，如果你想修改此配置，请确保此值小于canal端的canal.instance.memory.buffer.size配置（默认16384），否则在ack前Fe会尝试获取比store队列长度更多的event，导致store队列阻塞至超时为止。
min_bytes_sync_commit
提交事务需满足的最小数据大小。若Fe接收到的数据大小小于它，会继续等待下一批数据直到时间超过了sync_commit_interval_second为止。默认值是15MB，如果你想修改此配置，请确保此值小于canal端的canal.instance.memory.buffer.size和canal.instance.memory.buffer.memunit的乘积（默认16MB），否则在ack前Fe会尝试获取比store空间更大的数据，导致store队列阻塞至超时为止。
max_bytes_sync_commit
提交事务时的数据大小的最大值。若Fe接收到的数据大小大于它，会立即提交事务并发送已积累的数据。默认值是64MB，如果你想修改此配置，请确保此值大于canal端的canal.instance.memory.buffer.size和canal.instance.memory.buffer.memunit的乘积（默认16MB）和min_bytes_sync_commit。
max_sync_task_threads_num
数据同步作业线程池中的最大线程数量。此线程池整个FE中只有一个，用于处理FE中所有数据同步作业向BE发送数据的任务task，线程池的实现在SyncTaskPool类。
#常见问题
修改表结构是否会影响数据同步作业？
会影响。数据同步作业并不能禁止alter table的操作，当表结构发生了变化，如果列的映射无法匹配，可能导致作业发生错误暂停，建议通过在数据同步作业中显式指定列映射关系，或者通过增加 Nullable 列或带 Default 值的列来减少这类问题。
删除了数据库后数据同步作业还会继续运行吗？
不会。删除数据库后的几秒日志中可能会出现找不到元数据的错误，之后该数据同步作业会被FE的定时调度检查时停止。
多个数据同步作业可以配置相同的ip:port + destination吗？
不能。创建数据同步作业时会检查ip:port + destination与已存在的作业是否重复，防止出现多个作业连接到同一个instance的情况。
为什么数据同步时浮点类型的数据精度在Mysql端和Doris端不一样？
Doris本身浮点类型的精度与Mysql不一样。可以选择用Decimal类型代替。
