CREATE SYNC JOB

description

数据同步(Sync Job)功能，支持用户提交一个常驻的数据同步作业，通过从指定的远端地址读取Binlog日志，增量同步用户在Mysql数据库的对数据更新操作的CDC(Change Data Capture)功能。

目前数据同步作业只支持对接Canal，从Canal Server上获取解析好的Binlog数据，导入到Doris内。

用户可通过 SHOW SYNC JOB 查看数据同步作业状态。

语法：

1. CREATE SYNC [db.]job_name
2. (
3. channel_desc,
4. channel_desc
5. ...
6. )
7. binlog_desc
  1. job_name

同步作业名称，是作业在当前数据库内的唯一标识，相同job_name的作业只能有一个在运行。 2. channel_desc

作业下的数据通道，用来描述mysql源表到doris目标表的映射关系。

语法：

1. FROM mysql_db.src_tbl INTO des_tbl 2. [partitions] 3. [columns_mapping]

  1. mysql_db.src_tbl

    指定mysql端的数据库和源表。 2. des_tbl

    指定doris端的目标表，只支持Unique表，且需开启表的batch delete功能(开启方法请看help alter table的’批量删除功能’)。 3. partitions

    指定导入目的表的哪些 partition 中。如果不指定，则会自动导入到对应的 partition 中。

    示例：

    1. PARTITION(p1, p2, p3) 4. column_mapping

    指定mysql源表和doris目标表的列之间的映射关系。如果不指定，FE会默认源表和目标表的列按顺序一一对应。

    不支持 col_name = expr 的形式表示列。

    示例：

    ``` 1. 假设目标表列为(k1, k2, v1)，

    1. 改变列k1和k2的顺序

    2. COLUMNS(k2, k1, v1)

    3. 忽略源数据的第四列

    4. COLUMNS(k2, k1, v1, dummy_column) ```
    5. binlog_desc

用来描述远端数据源，目前仅支持canal一种。

语法：

1. FROM BINLOG 2. ( 3. "key1" = "value1", 4. "key2" = "value2" 5. )

  1. Canal 数据源对应的属性，以canal.为前缀

    1. canal.server.ip: canal server的地址
    2. canal.server.port: canal server的端口
    3. canal.destination: instance的标识
    4. canal.batchSize: 获取的batch大小的最大值，默认8192
    5. canal.username: instance的用户名
    6. canal.password: instance的密码
    7. canal.debug: 可选，设置为true时，会将batch和每一行数据的详细信息都打印出来

example

  1. 简单为 test_db 的 test_tbl 创建一个名为 job1 的数据同步作业，连接本地的Canal服务器，对应Mysql源表 mysql_db1.tbl1。

1. CREATE SYNC `test_db`.`job1` 2. ( 3. FROM `mysql_db1`.`tbl1` INTO `test_tbl ` 4. ) 5. FROM BINLOG 6. ( 7. "type" = "canal", 8. "canal.server.ip" = "127.0.0.1", 9. "canal.server.port" = "11111", 10. "canal.destination" = "example", 11. "canal.username" = "", 12. "canal.password" = "" 13. ); 2. 为 test_db 的多张表创建一个名为 job1 的数据同步作业，一一对应多张Mysql源表，并显式的指定列映射。

1. CREATE SYNC `test_db`.`job1` 2. ( 3. FROM `mysql_db`.`t1` INTO `test1` COLUMNS(k1, k2, v1) PARTITIONS (p1, p2), 4. FROM `mysql_db`.`t2` INTO `test2` COLUMNS(k3, k4, v2) PARTITION p1 5. ) 6. FROM BINLOG 7. ( 8. "type" = "canal", 9. "canal.server.ip" = "xx.xxx.xxx.xx", 10. "canal.server.port" = "12111", 11. "canal.destination" = "example", 12. "canal.username" = "username", 13. "canal.password" = "password" 14. );

keyword

1. CREATE,SYNC,JOB,BINLOG