Flink Mysql CDC結(jié)合Doris flink connector實現(xiàn)數(shù)據(jù)實時入庫
Apache doris通過擴展支持通過 Flink 讀寫 doris 數(shù)倉中的數(shù)據(jù)表，

目前 doris 支持 Flink 1.11.x ，1.12.x，1.13.x，Scala版本：2.12.x

目前Flink doris connector目前控制入庫通過兩個參數(shù)：

sink.batch.size ：每多少條寫入一次，默認100條
sink.batch.interval ：每個多少秒寫入一下，默認1秒
這兩參數(shù)同時起作用，那個條件先到就觸發(fā)寫doris表操作，

注意：

這里注意的是要啟用 http v2 版本，具體在 fe.conf 中配置 enable_http_server_v2=true，同時因為是通過 fe http rest api 獲取 be 列表，這倆需要配置的用戶有 admin 權限。

Flink Doris Connector 編譯

在 doris 的 docker 編譯環(huán)境 apache/incubator-doris:build-env-1.2 下進行編譯，因為 1.3 下面的JDK 版本是 11，會存在編譯問題。

在 extension/flink-doris-connector/ 源碼目錄下執(zhí)行：

sh build.sh
編譯成功后，會在 output/ 目錄下生成文件 doris-flink-1.0.0-SNAPSHOT.jar。將此文件復制到 Flink 的 ClassPath 中即可使用 Flink-Doris-Connector。例如，Local 模式運行的 Flink，將此文件放入 jars/ 文件夾下。Yarn集群模式運行的Flink，則將此文件放入預部署包中。

針對Flink 1.13.x版本適配問題

<properties>
<scala.version>2.12</scala.version>
<flink.version>1.11.2</flink.version>
<libthrift.version>0.9.3</libthrift.version>
<arrow.version>0.15.1</arrow.version>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<doris.home>{basedir}/../../thirdparty</doris.thirdparty>
</properties>
只需要將這里的 flink.version 改成和你 Flink 集群版本一致，重新編輯即可

使用示例

通過flink cdc實現(xiàn)mysql binlog日志數(shù)據(jù)的消費，然后通過flink doris connector sql實時導入mysql數(shù)據(jù)到doris表數(shù)據(jù)中

這個代碼已經(jīng)提交到apache doris的示例代碼庫里

org.apache.doris.demo.flink.FlinkConnectorMysqlCDCDemo
注意: 由于Flink doris connector jar包不在Maven中央倉庫中，需要單獨編譯并添加到你項目的classpath中。參考Flink doris connector的編譯和使用: Flink doris connector

首先Mysql 要開啟 binlog 具體如何打開binlog請自行搜索或到Mysql官方文檔查詢
安裝Flink，F(xiàn)link的安裝和使用這里不做介紹，只是在開發(fā)環(huán)境中給出代碼示例
創(chuàng)建Mysql數(shù)據(jù)庫表

 CREATE TABLE `test` (
  `id` int NOT NULL AUTO_INCREMENT,
  `name` varchar(255) DEFAULT NULL,
  PRIMARY KEY (`id`)
 ) ENGINE=InnoDB AUTO_INCREMENT=19 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci

創(chuàng)建doris表

CREATE TABLE `doris_test` (
  `id` int NULL COMMENT "",
  `name` varchar(100) NULL COMMENT ""
 ) ENGINE=OLAP
 DUPLICATE KEY(`id`)
 COMMENT "OLAP"
 DISTRIBUTED BY HASH(`id`) BUCKETS 1
 PROPERTIES (
 "replication_num" = "3",
 "in_memory" = "false",
 "storage_format" = "V2"
 );

創(chuàng)建Flink Mysql CDC

tEnv.executeSql(
  "CREATE TABLE orders (\n" +
  "  id INT,\n" +
  "  name STRING\n" +
  ") WITH (\n" +
  "  'connector' = 'mysql-cdc',\n" +
  "  'hostname' = 'localhost',\n" +
  "  'port' = '3306',\n" +
  "  'username' = 'root',\n" +
  "  'password' = 'zhangfeng',\n" +
  "  'database-name' = 'demo',\n" +
  "  'table-name' = 'test'\n" +
  ")");

創(chuàng)建Flink Doris Table 映射表

 tEnv.executeSql(
  "CREATE TABLE doris_test_sink (" +
  "id INT," +
  "name STRING" +
  ") " +
  "WITH (\n" +
  "  'connector' = 'doris',\n" +
  "  'fenodes' = '10.220.146.10:8030',\n" +
  "  'table.identifier' = 'test_2.doris_test',\n" +
  "  'sink.batch.size' = '2',\n" +
  "  'username' = 'root',\n" +
  "  'password' = ''\n" +
  ")");

執(zhí)行插入操作

 tEnv.executeSql("INSERT INTO doris_test_sink select id,name from orders");