Dinky 集成 Doris FlinkCDCTools 数据同步

Dinky开源 2024-09-29

203

导读：本文介绍了 Dinky 实时计算平台集成 Doris FlinkCDCTools 进行 Doris 整库同步。

社区公告：

问题反馈、代码提交、文章投稿与社区贡献请移步 Github issue。

Github issue #66 登记企业或组织生产使用，可邀请至企业支持群，获取团队技术支持与其他企业用户的经验分享。

GitHub 地址

https://github.com/DataLinkDC/dinky

https://gitee.com/DataLinkDC/Dinky

欢迎大家为 Dinky 送上小星星~

一、前言

Dinky 是一个基于 Apache Flink 的实时计算平台，它提供了一站式的 Flink 任务开发、运维、监控等功能。本教程一步一步的教你如何使用Dinky 运行 Doris FlinkCDCTools 任务实现整库同步 Doris 并自动建表功能。

二、Docker 快速部署 Dinky

h2数据库无需安装，开箱即用，仅适合快速体验，重启后数据消失，如要正式使用，请切换为mysql。

docker run --restart=always -p 8888:8888 \
  -v opt/lib:/opt/dinky/customJar/ \
  --name dinky \
  dinkydocker/dinky-standalone-server:1.1.0-flink1.19
复制

这里我们外挂了一个 opt/lib 目录，这是专门用来放依赖的，后面我们下载的doris和cdc依赖都放到这个目录下面。

三、快速部署 Doris 与 Mysql 测试环境

Flink CDC为我们提供了可快速部署的docker-compose yaml文件，我们可以很方便的创建一个测试环境出来。

如果你Mysql与Doris环境都已经具备，那么可以跳过此章节。

version: '2.1'
services:
  doris:
    image: yagagagaga/doris-standalone
    ports:
      - "8030:8030"
      - "8040:8040"
      - "9030:9030"
  mysql:
    image: debezium/example-mysql:1.1
    ports:
      - "3306:3306"
    environment:
      - MYSQL_ROOT_PASSWORD=123456
      - MYSQL_USER=mysqluser
      - MYSQL_PASSWORD=mysqlpw
复制

在 docker-compose.yml 所在目录下执行下面的命令来启动本教程需要的组件：

docker-compose up -d
复制

该命令将以 detached 模式自动启动 Docker Compose 配置中定义的所有容器。你可以通过 docker ps 来观察上述的容器是否正常启动了，也可以通过访问http://localhost:8030/ 来查看 Doris 是否运行正常。

四、准备数据

进入 MySQL 容器。

docker-compose exec mysql mysql -uroot -p123456
复制

创建数据库 app_db 和表 orders,products,shipments，并插入数据。

-- 创建数据库
CREATE DATABASE app_db;
USE app_db;
-- 创建 orders 表
CREATE TABLE `orders` (
`id` INT NOT NULL,
`price` DECIMAL(10,2) NOT NULL,
PRIMARY KEY (`id`)
);


-- 插入数据
INSERT INTO `orders` (`id`, `price`) VALUES (1, 4.00);
INSERT INTO `orders` (`id`, `price`) VALUES (2, 100.00);


复制

Doris 暂时不支持自动创建数据库，需要先创建写入表对应的数据库。进入 Doris Web UI。http://localhost:8030/默认的用户名为 root，默认密码为空。通过 Web UI 创建 app_db 数据库。

create database app_db;
复制

五、下载 Doris 依赖

本教程使用Flink 1.19环境，请根据你自己环境下载正确依赖版本。

地址：https://repo.maven.apache.org/maven2/org/apache/doris/

doris依赖 flink-doris-connector-1.19-24.0.0.jar

cdc依赖 flink-sql-connector-mysql-cdc-3.2.0.jar

mysql依赖 mysql-connector-j-8.2.0.jar

上述依赖下载完成后，除了flink-doris-connector-1.19-24.0.0.jar以外，其余jar包放到dinky的依赖目录下（dinky/extends 或者 docker部署的customJar下面），记得flink集群也要放。之后重启 Dinky 与 Flink。

六、注册 Flink 集群与 Jar 包

在注册中心，注册你的Flink集群。

点击资源，上传Doris Jar包。

检查jar是否存在，不存在则点一下同步目录结构即可。

七、开始运行

打开Dinky页面，新建Flink Jar任务。（重点！一定要建立 Flink Jar任务，千万不要建立Flink Sql任务）

输入以下代码,注意把相关IP替换成你自己的，Flink集群需要自己提前注册好，选择对应集群。

-- checkpoint设置，不写会出现数据无法写入情况
SET 'execution.checkpointing.interval' = '30s';


EXECUTE JAR WITH (
  -- 这里对应的就是我们刚刚在资源上传的Jar包
'uri'='rs:///flink-doris-connector-1.19-24.0.0.jar',
'main-class'='org.apache.doris.flink.tools.cdc.CdcTools',
'args'='    mysql-sync-database
    --database app_db
    --mysql-conf hostname=192.168.1.68
    --mysql-conf port=3306
    --mysql-conf username=root
    --mysql-conf password=123456
    --mysql-conf database-name=app_db
    --mysql-conf server-time-zone=UTC
    --including-tables "orders"
    --sink-conf fenodes=192.168.1.68:8030
    --sink-conf username=root
    --sink-conf jdbc-url=jdbc:mysql://192.168.1.68:9030
    --sink-conf sink.label-prefix=label-2
    --table-conf replication_num=1 '
);
复制

具体参数说明可以前往Doris官网查看，此处不再做说明 https://doris.apache.org/zh-CN/docs/ecosystem/flink-doris-connector#%E4%BD%BF%E7%94%A8-flink-cdc-%E6%8E%A5%E5%85%A5%E5%A4%9A%E8%A1%A8%E6%88%96%E6%95%B4%E5%BA%93-%E6%94%AF%E6%8C%81-mysqloraclepostgresqlsqlservermongodb