Flink CDC 系列 - 实现 MySQL 数据实时写入 Apache Doris-阿里云开发者社区

开发者社区> 阿里云实时计算Flink> 正文

Flink CDC 系列 - 实现 MySQL 数据实时写入 Apache Doris

简介: Flink CDC 结合 Doris Flink Connector 实现 MySQL 数据实时入 Apache Doris。
+关注继续查看

本文通过实例来演示怎么通过 Flink CDC 结合 Doris 的 Flink Connector 实现从 Mysql 数据库中监听数据并实时入库到 Doris 数仓对应的表中。主要内容包括:

  1. 什么是 CDC
  2. Flink CDC
  3. 什么是 Flink Doris Connector
  4. 用法示例

Flink 中文学习网站
https://flink-learning.org.cn

一、什么是 CDC

CDC 是变更数据捕获 (Change Data Capture) 技术的缩写,它可以将源数据库 (Source) 的增量变动记录,同步到一个或多个数据目的 (Sink)。在同步过程中,还可以对数据进行一定的处理,例如分组 (GROUP BY)、多表的关联 (JOIN) 等。

例如对于电商平台,用户的订单会实时写入到某个源数据库;A 部门需要将每分钟的实时数据简单聚合处理后保存到 Redis 中以供查询,B 部门需要将当天的数据暂存到 Elasticsearch 一份来做报表展示,C 部门也需要一份数据到 ClickHouse 做实时数仓。随着时间的推移,后续 D 部门、E 部门也会有数据分析的需求,这种场景下,传统的拷贝分发多个副本方法很不灵活,而 CDC 可以实现一份变动记录,实时处理并投递到多个目的地。

CDC 的应用场景

  • 数据同步:用于备份,容灾;
  • 数据分发:一个数据源分发给多个下游系统;
  • 数据采集:面向数据仓库 / 数据湖的 ETL 数据集成,是非常重要的数据源。

CDC 的技术方案非常多,目前业界主流的实现机制可以分为两种:

  • 基于查询的 CDC

    • 离线调度查询作业,批处理。把一张表同步到其他系统,每次通过查询去获取表中最新的数据;
    • 无法保障数据一致性,查的过程中有可能数据已经发生了多次变更;
    • 不保障实时性,基于离线调度存在天然的延迟。
  • 基于日志的 CDC

    • 实时消费日志,流处理,例如 MySQL 的 binlog 日志完整记录了数据库中的变更,可以把 binlog 文件当作流的数据源;
    • 保障数据一致性,因为 binlog 文件包含了所有历史变更明细;
    • 保障实时性,因为类似 binlog 的日志文件是可以流式消费的,提供的是实时数据。

二、Flink CDC

Flink 在 1.11 版本中新增了 CDC 的特性,简称改变数据捕获。名称来看有点乱,我们先从之前的数据架构来看 CDC 的内容。

img

以上是之前的 mysq binlog 日志处理流程,例如 canal 监听 binlog 把日志写入到 kafka 中。而 Apache Flink 实时消费 Kakfa 的数据实现 mysql 数据的同步或其他内容等。拆分来说整体上可以分为以下几个阶段:

  1. Mysql 开启 binlog;
  2. Canal 同步 binlog 数据写入到 Kafka;
  3. Flink 读取 Kakfa 中的 binlog 数据进行相关的业务处理。

整体的处理链路较长,需要用到的组件也比较多。Apache Flink CDC 可以直接从数据库获取到 binlog 供下游进行业务计算分析

Flink Connector Mysql CDC 2.0 特性

提供 MySQL CDC 2.0,核心 feature 包括:

  • 并发读取,全量数据的读取性能可以水平扩展;
  • 全程无锁,不对线上业务产生锁的风险;
  • 断点续传,支持全量阶段的 checkpoint。

网上有测试文档显示用 TPC-DS 数据集中的 customer 表进行了测试,Flink 版本是 1.13.1,customer 表的数据量是 6500 万条,Source 并发为 8,全量读取阶段:

  • MySQL CDC 2.0 用时 13 分钟;
  • MySQL CDC 1.4 用时 89 分钟;
  • 读取性能提升 6.8 倍。

三、什么是 Flink Doris Connector

Flink Doris Connector 是 Doris 社区为了方便用户使用 Flink 读写 Doris 数据表的一个扩展,目前 Doris 支持 Flink 1.11.x ,1.12.x,1.13.x;Scala 版本:2.12.x。

目前 Flink Doris connector 目前控制入库通过两个参数:

  1. sink.batch.size:每多少条写入一次,默认 100 条;
  2. sink.batch.interval :每个多少秒写入一下,默认 1 秒。

这两参数同时起作用,哪个条件先到就触发写 Doris 表操作,

注意:

这里注意的是要启用 http v2 版本,具体在 fe.conf 中配置 enable_http_server_v2=true,同时因为是通过 fe http rest api 获取 be 列表,这俩需要配置的用户有 admin 权限。

四、用法示例

4.1 Flink Doris Connector 编译

首先我们要编译 Doris 的 Flink connector,也可以通过下面的地址进行下载:

https://github.com/hf200012/hf200012.github.io/raw/main/lib/doris-flink-1.0-SNAPSHOT.jar

注意:

这里因为 Doris 的 Flink Connector 是基于 Scala 2.12.x 版本进行开发的,所以你在使用 Flink 的时候请选择对应 Scala 2.12 的版本,如果你使用上面地址下载了相应的 jar,请忽略下面的编译内容部分。

在 Doris 的 docker 编译环境 apache/incubator-doris:build-env-1.2 下进行编译,因为 1.3 下面的 JDK 版本是 11,会存在编译问题。

在 extension/flink-doris-connector/ 源码目录下执行:

sh build.sh

编译成功后,会在 output/ 目录下生成文件 doris-flink-1.0.0-SNAPSHOT.jar。将此文件复制到 FlinkClassPath 中即可使用 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}/../../</doris.home>
        <doris.thirdparty>${basedir}/../../thirdparty</doris.thirdparty>
    </properties>

只需要将这里的 flink.version 改成和你 Flink 集群版本一致,重新编辑即可。

4.2 配置 Flink

这里我们是通过 Flink Sql Client 方式来进行操作。

这里我们演示使用的软件版本:

  1. Mysql 8.x
  2. Apache Flink :1.13.3
  3. Apache Doris :0.14.13.1

4.2.1 安装 Flink

首先下载和安装 Flink :

https://dlcdn.apache.org/flink/flink-1.13.3/flink-1.13.3-bin-scala_2.12.tgz

这里演示使用的是本地单机模式:

# wget https://dlcdn.apache.org/flink/flink-1.12.5/flink-1.12.5-bin-scala_2.12.tgz
# tar zxvf flink-1.12.5-bin-scala_2.12.tgz 

下载 Flink CDC 相关 Jar 包:

https://repo1.maven.org/maven2/com/ververica/flink-connector-mysql-cdc/2.0.2/flink-connector-mysql-cdc-2.0.2.jar

这里注意 Flink CDC 和 Flink 的版本对应关系。

image-20211025170642628

  • 将上面下载或者编译好的 Flink Doris Connector jar 包复制到 Flink 根目录下的 lib 目录下;
  • Flink CDC 的 jar 包也复制到 Flink 根目录下的 lib 目录下。

image-20211026095513892

4.2.2 启动 Flink

这里我们使用的是本地单机模式。

# bin/start-cluster.sh
Starting cluster.
Starting standalonesession daemon on host doris01.
Starting taskexecutor daemon on host doris01.

我们通过 web 访问 (默认端口是 8081) 启动起来 Flink 集群,可以看到集群正常启动。

image-20211025162831632

4.3 安装 Apache Doris

具体安装部署 Doris 的方法,参照下面的连接:

https://hf200012.github.io/2021/09/Apache-Doris-环境安装部署

4.4 安装配置 Mysql

  1. 安装 Mysql,快速使用 Docker 安装配置 Mysql,具体参照下面的连接:

    https://segmentfault.com/a/1190000021523570

  2. 开启 Mysql binlog,进入 Docker 容器修改 /etc/my.cnf 文件,在 [mysqld] 下面添加以下内容,

    log_bin=mysql_bin
    binlog-format=Row
    server-id=1

    然后重启 Mysql。

    systemctl restart mysqld
  3. 创建 Mysql 数据库表。
 CREATE TABLE `test_cdc` (
  `id` int NOT NULL AUTO_INCREMENT,
  `name` varchar(255) DEFAULT NULL,
  PRIMARY KEY (`id`)
 ) ENGINE=InnoDB 

4.5 创建 Doris 表

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

4.6 启动 Flink Sql Client

./bin/sql-client.sh embedded
> set execution.result-mode=tableau;

image-20211025165547903

4.6.1 创建 Flink CDC Mysql 映射表

CREATE TABLE test_flink_cdc ( 
  id INT, 
  name STRING,
  primary key(id)  NOT ENFORCED
) WITH ( 
  'connector' = 'mysql-cdc', 
  'hostname' = 'localhost', 
  'port' = '3306', 
  'username' = 'root', 
  'password' = 'password', 
  'database-name' = 'demo', 
  'table-name' = 'test_cdc' 
);

执行查询创建的 Mysql 映射表,显示正常。

select * from test_flink_cdc;

image-20211026100505972

4.6.2 创建 Flink Doris Table 映射表

使用 Doris Flink Connector 创建 Doris 映射表。

CREATE TABLE doris_test_sink (
   id INT,
   name STRING
) 
WITH (
  'connector' = 'doris',
  'fenodes' = 'localhost:8030',
  'table.identifier' = 'db_audit.doris_test',
  'sink.batch.size' = '2',
  'sink.batch.interval'='1',
  'username' = 'root',
  'password' = ''
)

在命令行下执行上面的语句,可以看到创建表成功,然后执行查询语句,验证是否正常。

select * from doris_test_sink;

image-20211026100804091

执行插入操作,将 Mysql 里的数据通过 Flink CDC 结合 Doris Flink Connector 方式插入到 Doris 中。

INSERT INTO doris_test_sink select id,name from test_flink_cdc

image-20211026101004547

提交成功之后我们在 Flink 的 Web 界面可以看到相关的 Job 任务信息。

image-20211026100943474

4.6.3 向 Mysql 表中插入数据

INSERT INTO test_cdc VALUES (123, 'this is a update');
INSERT INTO test_cdc VALUES (1212, '测试flink CDC');
INSERT INTO test_cdc VALUES (1234, '这是测试');
INSERT INTO test_cdc VALUES (11233, 'zhangfeng_1');
INSERT INTO test_cdc VALUES (21233, 'zhangfeng_2');
INSERT INTO test_cdc VALUES (31233, 'zhangfeng_3');
INSERT INTO test_cdc VALUES (41233, 'zhangfeng_4');
INSERT INTO test_cdc VALUES (51233, 'zhangfeng_5');
INSERT INTO test_cdc VALUES (61233, 'zhangfeng_6');
INSERT INTO test_cdc VALUES (71233, 'zhangfeng_7');
INSERT INTO test_cdc VALUES (81233, 'zhangfeng_8');
INSERT INTO test_cdc VALUES (91233, 'zhangfeng_9');

4.6.4 观察 Doris 表的数据

首先停掉 Insert into 这个任务,因为我是在本地单机模式,只有一个 task 任务,所以要停掉,然后在命令行执行查询语句才能看到数据。

image-20211026101203629

4.6.5 修改 Mysql 的数据

重新启动 Insert into 任务:

image-20211025182341086

修改 Mysql 表里的数据:

update test_cdc set name='这个是验证修改的操作' where id =123

再去观察 Doris 表中的数据,你会发现已经修改。

注意这里如果要想 Mysql 表里的数据修改,Doris 里的数据也同样修改,Doris 数据表的模型要是 Unique key 模型,其他数据模型 (Aggregate Key 和 Duplicate Key) 不能进行数据的更新操作。

image-20211025182435827

4.6.6 删除数据操作

目前 Doris Flink Connector 还不支持删除操作,后面计划会加上这个操作。

更多 Flink CDC 相关技术问题,可扫码加入社区钉钉交流群~

img


相关文章

近期热点

img


更多 Flink 相关技术问题,可扫码加入社区钉钉交流群
第一时间获取最新技术文章和社区动态,请关注公众号~

image.png

活动推荐

阿里云基于 Apache Flink 构建的企业级产品-实时计算Flink版现开启活动:
99 元试用 实时计算Flink版(包年包月、10CU)即有机会获得 Flink 独家定制卫衣;另包 3 个月及以上还有 85 折优惠!
了解活动详情:https://www.aliyun.com/product/bigdata/sc

image.png

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
Apache Flink 在实时金融数据湖的应用
本文由中原银行大数据平台研发工程师白学余分享,主要介绍实时金融数据湖在中原银行的应用。
1967 0
阿里云服务器怎么设置密码?怎么停机?怎么重启服务器?
如果在创建实例时没有设置密码,或者密码丢失,您可以在控制台上重新设置实例的登录密码。本文仅描述如何在 ECS 管理控制台上修改实例登录密码。
9485 0
趣头条基于 Flink+ClickHouse 构建实时数据分析平台
本文由趣头条数据平台负责人王金海分享,主要介绍趣头条 Flink-to-Hive 小时级场景和 Flink-to-ClickHouse 秒级场景。
3749 0
阿里云服务器端口号设置
阿里云服务器初级使用者可能面临的问题之一. 使用tomcat或者其他服务器软件设置端口号后,比如 一些不是默认的, mysql的 3306, mssql的1433,有时候打不开网页, 原因是没有在ecs安全组去设置这个端口号. 解决: 点击ecs下网络和安全下的安全组 在弹出的安全组中,如果没有就新建安全组,然后点击配置规则 最后如上图点击添加...或快速创建.   have fun!  将编程看作是一门艺术,而不单单是个技术。
10842 0
阿里云服务器如何登录?阿里云服务器的三种登录方法
购买阿里云ECS云服务器后如何登录?场景不同,阿里云优惠总结大概有三种登录方式: 登录到ECS云服务器控制台 在ECS云服务器控制台用户可以更改密码、更换系.
13168 0
MySQL---数据库从入门走向大神系列(十一)-Java获取数据库/结果集的元信息、将数据表写入excel表格
数据库的元信息: 首先介绍一下数据库的元信息(元数据): 元数据(Metadata)是关于数据的数据。 元数据是描述数据仓库内数据的结构和建立方法的数据。 存储的数据是什么类型,什么驱动等等,这些描述数据的数据,就是元数据! 准备: package cn.
1294 0
MySQL---数据库从入门走向大神系列(九)-用Java向数据库读写大文本/二进制文件数据
介绍MySQL的文本和图形数据类型: Text 类型: 数据类型:描述 ------------------------------------------------------ char(size):保存固定长度的字符串(可包含字母、数字以及特殊字符)。
995 0
阿里云ECS云服务器初始化设置教程方法
阿里云ECS云服务器初始化是指将云服务器系统恢复到最初状态的过程,阿里云的服务器初始化是通过更换系统盘来实现的,是免费的,阿里云百科网分享服务器初始化教程: 服务器初始化教程方法 本文的服务器初始化是指将ECS云服务器系统恢复到最初状态,服务器中的数据也会被清空,所以初始化之前一定要先备份好。
6886 0
基于 Flink SQL CDC 的实时数据同步方案
Flink 1.11 引入了 Flink SQL CDC,CDC 能给我们数据和业务间能带来什么变化?本文由 Apache Flink PMC,阿里巴巴技术专家伍翀 (云邪)分享,内容将从传统的数据同步方案,基于 Flink CDC 同步的解决方案以及更多的应用场景和 CDC 未来开发规划等方面进行介绍和演示。
23223 0
593
文章
18
问答
来源圈子
更多
实时计算 Flink 版(Alibaba Cloud Realtime Compute for Apache Flink,Powered by Ververica)是阿里云基于 Apache Flink 构建的企业级、高性能实时大数据处理系统,由 Apache Flink 创始团队官方出品,拥有全球统一商业化品牌,完全兼容开源 Flink API,提供丰富的企业级增值功能。
+ 订阅
文章排行榜
最热
最新
相关电子书
更多
《2021云上架构与运维峰会演讲合集》
立即下载
《零基础CSS入门教程》
立即下载
《零基础HTML入门教程》
立即下载