Doris：Binlog Load导入数据

news2024/11/23 7:32:17

Binlog Load提供了一种使Doris增量同步用户在Mysql数据库的对数据更新操作的CDC(Change Data Capture)功能。Binlog Load需要依赖canal作为中间媒介，让canal伪造成一个从节点去获取Mysql主节点上的Binlog并解析，再由Doris去获取Canal上解析好的数据。

1.适用场景

INSERT/UPDATE/DELETE支持
过滤Query
暂不兼容DDL语句
Binlog Load只能支持Unique类型的目标表，且必须激活目标表的Batch Delete功能。

2.配置Mysql端

2.1开启binlog日志功能

编辑my.cnf配置文件配置binlog日志信息：

[mysqld]
log-bin = mysql-bin # 开启 binlog
binlog-format=ROW # 选择 ROW 模式

######开启GTID模式 [可选]######
#gtid-mode=on // 开启gtid模式
#enforce-gtid-consistency=1 // 强制gtid和事务的一致性

2.2 配置授权账号

授权 canal 链接 MySQL 账号具有作为 MySQL slave 的权限, 如果已有账户可直接 grant。

create user canal@'%' identified with mysql_native_password by 'wsx-123';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
-- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ;
FLUSH PRIVILEGES;

3.配置Canal端

3.1下载 canal

wget https://github.com/alibaba/canal/releases/download/canal-1.1.6/canal.adapter-1.1.6.tar.gz

3.2 解压canal

mkdir ~/opt/canal

tar -zxf canal.deployer-1.1.6.tar.gz -C ~/opt/canal

3.3 配置canal

vi conf/example/instance.properties

3.4 启动canal

sh bin/startup.sh

3.5 验证启动成功

tail -f logs/example/example.log

4.基本操作

4.1 创建Doris目标表

 CREATE TABLE `dim_area_canal` (
  `id` bigint(20) NOT NULL,
  `level` tinyint(4) NOT NULL COMMENT '层级',
  `parent_code` varchar(20) NOT NULL DEFAULT "0" COMMENT '父级行政代码',
  `area_code` varchar(20) NOT NULL DEFAULT "0" COMMENT '行政代码',
  `name` varchar(255) NOT NULL DEFAULT "" COMMENT '名称',
  `short_name` varchar(255) NOT NULL DEFAULT "" COMMENT '简称',
  `merger_name` varchar(255) NOT NULL DEFAULT "" COMMENT '组合名',
  `pinyin` varchar(50) NOT NULL DEFAULT "" COMMENT '拼音',
  `lng` DECIMAL(10, 6) NOT NULL COMMENT '经度',
  `lat` DECIMAL(10, 6) NOT NULL COMMENT '纬度',
  `pcode` varchar(2) NULL,
  `ccode` varchar(2) NULL,
  `acode` varchar(2) NULL,
  `tcode` varchar(3) NULL,
  `vcode` varchar(3) NULL
) ENGINE=OLAP
UNIQUE KEY(`id`)
COMMENT 'OLAP'
DISTRIBUTED BY HASH(`id`) BUCKETS 10

！！Doris表结构和Mysql表结构字段顺序必须保持一致！！

4.2 开启Batch Delete

ALTER TABLE demo.dim_area_canal ENABLE FEATURE "BATCH_DELETE";

4.3 创建同步作业

CREATE SYNC `demo`.`dim_area_canal_job`
(
FROM `demo`.`dim_area` INTO `dim_area_canal`
(id,level,parent_code,area_code,name,short_name,merger_name,pinyin,lng,lat,pcode,ccode,acode,tcode,vcode)
)
FROM BINLOG
(
"type" = "canal",
"canal.server.ip" = "192.168.179.131",
"canal.server.port" = "11111",
"canal.destination" = "example",
"canal.username" = "canal",
"canal.password" = "canal"
);