mkdir -p /your_path/doris/jdbc_drivers 

jdbc_drivers_dir = /opt/doris/jdbc_drivers 

# 重啟FE 
cd /opt/doris/fe 
bin/stop_fe.sh && bin/start_fe.sh --daemon 
# 重啟BE 
cd /opt/doris/be 
bin/stop_be.sh && bin/start_be.sh --daemon 

CREATE CATALOG mysql PROPERTIES (
    "type"="jdbc",
    "user"="root",
    "password"="secret",
    "jdbc_url" = "jdbc:mysql://example.net:3306",
    "driver_url" = "mysql-connector-j-8.3.0.jar",
    "driver_class" = "com.mysql.cj.jdbc.Driver"
)

-- 直接查詢(xún)mysql表 
SELECT * FROM mysql_catalog.mysql_db.user_table LIMIT 10; 
-- 與Doris表關(guān)聯(lián)查詢(xún) 
SELECT a.id, a.name, b.sales 
FROM mysql_catalog.mysql_db.user_table a 
JOIN doris_table b ON a.id = b.user_id; 

-- 建表
create table as select  mysql_catalog.mysql_db.user_table
-- 導(dǎo)入
insert into doris_table select * from mysql_catalog.mysql_db.user_table

[mysqld] 
log-bin=mysql-bin         # 開(kāi)啟Binlog 
binlog-format=ROW          # 使用ROW模式 
binlog-row-image=FULL      # 記錄完整行數(shù)據(jù) 
server-id=1                # 唯一服務(wù)器ID 

sudo systemctl restart mysqld 

wget https://github.com/alibaba/canal/releases/download/canal-1.1.5/canal.deployer-1.1.5.tar.gz 
tar -zxvf canal.deployer-1.1.5.tar.gz 

mkdir conf/demo 
cp conf/example/instance.properties conf/demo/ 
vim conf/demo/instance.properties 

canal.instance.master.address=192.168.1.100:3306   # MySQL地址 
canal.instance.dbUsername=canal                  # 具有Binlog讀取權(quán)限的用戶(hù) 
canal.instance.dbPassword=canal                  # 用戶(hù)密碼 
canal.destination=demo                           # Instance名稱(chēng) 

sh bin/startup.sh 
# 驗(yàn)證日志：cat logs/demo/demo.log，顯示"start successful"表示啟動(dòng)成功 

--create Mysql table
CREATE TABLE `test.source_test` (
  `id` int(11) NOT NULL COMMENT "",
  `name` int(11) NOT NULL COMMENT ""
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
-- create Doris table
CREATE TABLE `target_test` (
  `id` int(11) NOT NULL COMMENT "",
  `name` int(11) NOT NULL COMMENT ""
) ENGINE=OLAP
UNIQUE KEY(`id`)
COMMENT "OLAP"
DISTRIBUTED BY HASH(`id`) BUCKETS 8
PROPERTIES (
"replication_allocation" = "tag.location.default: 3"
);

CREATE SYNC demo_job ( 
   FROM mysql.test_table INTO target_test (id, name) 
) 
FROM BINLOG ( 
   "type" = "canal", 
   "canal.server.ip" = "192.168.1.101",   # Canal服務(wù)器IP 
   "canal.server.port" = "11111",         # Canal默認(rèn)端口 
   "canal.destination" = "demo",          # Canal Instance名稱(chēng) 
   "canal.username" = "canal", 
   "canal.password" = "canal", 
   "sync.mode" = "incremental"            # 增量同步模式 
); 

CREATE USER 'canal'@'%' IDENTIFIED BY 'canal'; 
GRANT REPLICATION SLAVE ON *.* TO 'canal'@'%'; 
FLUSH PRIVILEGES; 

<dependency>
  <groupId>org.apache.doris</groupId>
  <artifactId>flink-doris-connector-1.16</artifactId>
  <version>25.1.0</version>
</dependency> 

-- enable checkpoint
SET 'execution.checkpointing.interval' = '30s';
-- 創(chuàng)建MySQL CDC源表 
CREATE TABLE mysql_cdc_source ( 
   id INT PRIMARY KEY NOT ENFORCED, 
   name STRING, 
   age INT, 
   ts TIMESTAMP(3) METADATA FROM 'source_ts' VIRTUAL 
) WITH ( 
   'connector' = 'mysql-cdc', 
   'hostname' = '192.168.1.100', 
   'port' = '3306', 
   'username' = 'root', 
   'password' = 'your_mysql_password', 
   'database-name' = 'test', 
   'table-name' = 'user_table' 
); 
-- 創(chuàng)建Doris目標(biāo)表 
CREATE TABLE doris_sink ( 
   id INT, 
   name STRING, 
   age INT, 
   update_time TIMESTAMP(3) 
) WITH ( 
   'connector' = 'doris', 
   'fenodes' = 'doris-fe:8030', 
   'table.identifier' = 'test.user_table', 
   'username' = 'root', 
   'password' = '', 
   'sink.properties.format' = 'json', 
   'sink.enable-delete' = 'true', 
   'sink.label-prefix' = 'flink_cdc_' 
); 
-- 啟動(dòng)數(shù)據(jù)同步 
INSERT INTO doris_sink SELECT id, name, age, ts FROM mysql_cdc_source; 

<FLINK_HOME>bin/flink run \
    -Dexecution.checkpointing.interval=10s \
    -Dparallelism.default=1 \
    -c org.apache.doris.flink.tools.cdc.CdcTools \
    lib/flink-doris-connector-1.16-24.0.1.jar \
    mysql-sync-database \
    --database test_db \
    --mysql-conf hostname=127.0.0.1 \
    --mysql-conf port=3306 \
    --mysql-conf username=root \
    --mysql-conf password=123456 \
    --mysql-conf database-name=mysql_db \
    --including-tables "tbl1|test.*" \
    --sink-conf fenodes=127.0.0.1:8030 \
    --sink-conf username=root \
    --sink-conf password=123456 \
    --sink-conf jdbc-url=jdbc:mysql://127.0.0.1:9030 \
    --sink-conf sink.label-prefix=label \
    --table-conf replication_num=1 

# 下載Datax壓縮包
wget https://datax-opensource.oss-cn-hangzhou.aliyuncs.com/202308/datax.tar.gz
tar -zxvf datax.tar.gz

{
    "job": {
        "content": [
            {
                "reader": {
                    "name": "mysqlreader",
                    "parameter": {
                        "column": [xxx],
                        "connection": [
                            {
                                "jdbcUrl": ["jdbc:mysql://localhost:3306/demo"],
                                "table": ["table_name"]
                            }
                        ],
                        "username": "root",
                        "password": "xxxxx",
                        "where": ""
                    }
                },
                "writer": {
                    "name": "doriswriter",
                    "parameter": {
                        "loadUrl": ["127.0.0.1:8030"],
                        "column": [xxxx],
                        "username": "root",
                        "password": "xxxxxx",
                        "postSql": ["select count(1) from table_name"],
                        "preSql": [],
                        "flushInterval":30000,
                        "connection": [
                          {
                            "jdbcUrl": "jdbc:mysql://127.0.0.1:9030/demo",
                            "selectedDatabase": "demo",
                            "table": ["table_name"]
                          }
                        ],
                        "loadProps": {
                            "format": "json",
                            "strip_outer_array":"true",
                            "line_delimiter": "\\x02"
                        }
                    }
                }
            }
        ],
        "setting": {
            "speed": {
                "channel": "1"
            }
        }
    }
}

python bin/datax.py mysql_to_doris.json