Flink 内容分享：Doris Connector 结合 Flink CDC 实现 MySQL 分库分表 Exactly Once精准接入

-- 引入Flink CDC的MySQL source connector
CREATE TABLE source_mysql (
  id INT,
  name STRING,
  amount BIGINT,
  ts TIMESTAMP(3),
  database_name STRING,
  table_name STRING,
  op_type STRING
) WITH (
 'connector' = 'mysql-cdc',
 'hostname' = 'your_mysql_host_ip',
 'port' = '3306',
 'username' = 'your_username',
 'password' = 'your_password',
 'debezium.snapshot.mode' = 'initial',
 'database-name' = 'your_database_name',
 'table-name' = 'your_table_name'
);
 
-- 定义Doris sink，使用Doris Connector
CREATE TABLE sink_doris (
  id INT,
  name STRING,
  amount BIGINT,
  ts TIMESTAMP(3),
  database_name STRING,
  table_name STRING,
  op_type STRING
) WITH (
  'connector' = 'doris',
  'fenodes' = 'fe_ip1:8030,fe_ip2:8030',
  'table.identifier' = 'db1.tbl1',
  'username' = 'your_username',
  'password' = 'your_password',
  'sink.batch.size' = '10000',
  'sink.batch.interval' = '2000',
  'sink.max-retries' = '3'
);
 
-- 将数据从MySQL source流式接入Doris sink
INSERT INTO sink_doris
SELECT * FROM source_mysql;

这个示例展示了如何使用Flink SQL DDL定义源表（source\_mysql）和目标表（sink\_doris），并通过INSERT语句实现从源表到目标表的数据流动。这个过程中，我们使用了Flink CDC连接器来捕获MySQL的变更数据，并且结合Doris Connector将数据写入Doris数据库。这个例子是实现MySQL分库分表数据同步到Doris的一个简化版本，但是它展示了如何将这些技术组合起来以解决实际的数据集成问题。