微信公众号里怎么做网站,河北省建设工程招标投标协会网站,wordpress分享插件带赞,北京有哪些电商平台公司面对客户环境中长期运行的各种类型的传统数据库#xff0c;如何优雅地设计数据迁移的方案#xff0c;既能灵活地应对各种数据导入场景和多源异构数据库#xff0c;又能满足客户对数据导入结果的准确性、一致性、实时性的要求#xff0c;让客户平滑地迁移到 PieCloudDB 数据…面对客户环境中长期运行的各种类型的传统数据库如何优雅地设计数据迁移的方案既能灵活地应对各种数据导入场景和多源异构数据库又能满足客户对数据导入结果的准确性、一致性、实时性的要求让客户平滑地迁移到 PieCloudDB 数据库生态是一个巨大的挑战。PieCloudDB Database 打造了丰富的数据同步工具来实现数据的高效流动本文将聚焦 PieCloudDB Flink Connector 工具进行详细的介绍。
拓数派旗下 PieCloudDB 是一款云原生分布式虚拟数仓为企业提供全新基于云数仓数字化解决方案助力企业建立以数据资产为核心的竞争壁垒以云资源最优化配置实现无限数据计算可能。PieCloudDB 通过多种创新性技术将物理数仓整合到云原生数据计算平台实现了分析型数据仓库上云虚拟化打造了存储计算分离的全新 eMPP 架构突破了传统 MPP 数据库多种瓶颈限制打破客户生产环境数据孤岛的同时也实现了按需瞬间扩缩容大大减少了存储空间的浪费。
Apache Flink 是一个分布式流计算处理引擎用于在无界或有界数据流上进行有状态的计算。它在所有的通用集群环境中都可以运行在任意规模下都可以达到内存级的计算速度。Flink 最初由德国柏林工业大学的 Stratosphere 项目发展而来是为了支持复杂的大规模数据分析任务而设计的并于2014年成为 Apache 软件基金会的顶级项目。用户可以运用 Flink 提供的 DataStream API 或 Table SQL API实现功能强大且高效的实时数据计算能力。此外Flink 原生支持的 checkpoint 机制可以为用户提供数据的一致性的保证。
Apache Flink 作为一个流处理框架与其他开源项目和工具的整合非常紧密。经过多年的发展整个 Flink 社区已经围绕 Flink 构成出了一个丰富的生态系统。PieCloudDB 组件 PieCloudDB Flink Connector 是拓数派团队自研的一款 Flink 连接器 可用于将来自 Flink 系统中的数据高效地写入 PieCloudDB配合 Flink 的 checkpoint 机制来保证数据导入结果的精准一次语义。本文将详细介绍 PieCloudDB Flink Connector 的功能和原理并结合实例进行演示。
1 PieCloudDB Flink Connector 功能介绍
PieCloudDB Flink Connector 可提供多种将 Flink 数据导入 PieCloudDB 的方式包括 Append-Only 模式和 Merge 模式以满足不同级别的导入语义。
在接入方式上PieCloudDB Flink Connector 提供多种选择包括使用 Flink DataStream API 编写相关的作业代码集成该组件或者直接利用 Flink SQL 语句使用该组件。
PieCloudDB Flink Connector 提供 Merge 导入模式采用幂等写方案配合 Flink 原生支持的 checkpoint 机制能够保证导入结果的可靠性和一致性。
此外PieCloudDB Flink Connector 不仅支持支持单表实时数据导入也可以支持整库实时数据同时导入。不过后者仅支持 Flink DataStream API 的接入方式不支持使用 Flink SQL 语法。
2 PieCloudDB Flink Connector 原理
2.1 精准一次导入原理
PieCloudDB Flink Connector 中的 PieCloudDBSink、PieCloudDBWriter类分别实现了 Flink 的 StatefulSink、StatefulSinkWriter 接口。
当开启 Flink 的 checkpoint 功能后在一个特定的 checkpoint 执行期间PieCloudDBWriter 负责将接收到的数据源源不断地写入内存管道同时异步线程会将这些数据拷贝到 PieCloudDB 中的内存临时表中。当 PieCloudDB Flink Connector 算子接收到 checkpoint 信号之后会先等待数据全部拷贝进 PieCloudDB 后再执行第二个阶段的动作包括数据的合并以及写入物理表。整个过程中一旦出现异常Flink 引擎就会自动从上一个 checkpoint 开始恢复作业保证不会发生数据丢失。
在第二个阶段中PieCloudDB Flink Connector 采用了幂等写方案来保证不会出现数据重复。具体做法是当数据全部导入 PieCloudDB 的临时表后根据表的主键字段以及数据的时序关系进行合并操作聚合这一段时间内每个主键对应记录的所有增删改操作得到最终结果只将该结果写入 PieCloudDB 中。
这里的聚合操作是先根据写入数据的主键和时序进行组合然后删除目标表中该主键对应的记录最后将最新的修改或新增写入目标表。比如一条主键为1的记录按照时间顺序发生了修改和删除操作那么最终结果是将该主键对应的记录从 PieCloudDB 中删除。数据的写入时序与数据在 Flink 中的顺序一致这些时序信息是通过在临时表扩充一个单独的 bigint 列来对每条数据做跟踪而记录下来的。应用这种幂等写方案可以确保即使发生了数据重复也能保证精确一次的导入语义。
上述 checkpoint 机制和幂等写过程如下图所示 checkpoint 机制及幂等写过程
2.2 整库同步原理
对于整库同步场景来说需要解决的是多表同时写入的问题和连接池通用性问题。 首先对于整库同步场景来说可能会存在多张表同时写入 PieCloudDB Flink Connector前面处理单表的逻辑可能会导致新表数据处理的不及时导致数据丢失。PieCloudDB Flink Connector 在内部维护了一套 Loader 池在同一个 Flink checkpoint 周期内会为每张表都分配一个对应的 Loader缓存在 Loader 池中。每张表写入的数据都分配给对应的 Loader 来处理这里的处理逻辑与单表导入的处理逻辑一致。唯一的不同点在于在 Flink checkpoint 周期结束时PieCloudDB Flink Connector 会将该 checkpoint 内所有存在数据写入的表都刷入 PieCloudDB 中。
另一个在整库同步场景需要解决的问题是连接池的通用性问题。根据上面的设计PieCloudDB Flink Connector 的每个实例都会维护一个 Loader 池每个 Loader 都会占用一个 PieCloudDB 数据库连接。如果用户需要提升导入性能一般最直接的做法是增加 Flink 作业的并行度即创建多个 PieCloudDB Flink Connector 的实例来加速整库同步的速度这对于具有海量历史数据的场景非常有必要。但这样会导致整个 Flink 作业需要的 PieCloudDB 数据库连接非常多且不可控因为无论是数据库表的数量还是作业的并行度都是无法预估的。仅仅通过设置内部连接池的最大连接数只会导致作业运行时由于获取不到新的连接而无法处理进而导致任务失败。
为了解决这一问题PieCloudDB Flink Connector 在内部设计了一套简易的排队算法如果数据库连接池的连接数已被占满那么单个 checkpoint 内新来的表需要进行排队直到 checkpoint 结束时等待其他表写入数据库并释放连接后才会将这些排队中的表导入数据库。在此期间这些排队的表的所有相关数据都会暂存到内存中。为了避免出现内存溢出这里的原始数据已经提前解析好只留下可以用于导入过程的关键信息以大大降低内存使用率避免出现内存溢出。使用此模型后整库同步的过程就能保证数据库连接数全程都是可控的最多不超过并发数和单个 PieCloudDB 数据库连接池最大连接数的乘积。
PieCloudDB Flink Connector 的整库同步功能需配合 PieCloudDB 动态作业执行器来使用后续的文章中会进一步描述欢迎关注
3 PieCloudDB Flink Connector 使用演示
接下来我们将用 MySQL 作为数据源来演示一下使用 PieCloudDB Flink Connector 将 MySQL 中的数据同步到 PieCloudDB 的过程。
创建 MySQL 源表
create table student (id int primary key, name varchar(32), score int, sex char(1)); mysql desc student;
-------------------------------------------------
| Field | Type | Null | Key | Default | Extra |
-------------------------------------------------
| id | int | NO | PRI | NULL | |
| name | varchar(32) | YES | | NULL | |
| score | int | YES | | NULL | |
| sex | char(1) | YES | | NULL | |
-------------------------------------------------
4 rows in set (0.03 sec)插入一些数据
insert into student (id, name, score, sex) values (1, student1, 65, 1);
insert into student (id, name, score, sex) values (2, student2, 75, 0);
insert into student (id, name, score, sex) values (3, student3, 85, 1);
insert into student (id, name, score, sex) values (4, student4, 95, 0);mysql select * from student;
---------------------------
| id | name | score | sex |
---------------------------
| 1 | student1 | 65 | 1 |
| 2 | student2 | 75 | 1 |
| 3 | student3 | NULL | NULL |
| 4 | student4 | NULL | NULL |
---------------------------
4 rows in set (0.01 sec)创建 PieCloudDB 目标表可以在 PieCloudDB 云原生管理平台的「数据洞察」功能页面进行
create table student (id int primary key, name varchar(32), score int, sex char(1)); demo \d studentTable public.studentColumn | Type | Collation | Nullable | Default
-------------------------------------------------------------
id | integer | | not null |
name | character varying(32) | | |
score | integer | | |
sex | character(1) | | |目前是一张空表
demo select * from student;id | name | score | sex
----------------------
(0 rows)启动 Flink 集群
ubuntu :: work/flink/flink-1.18.0 bin/start-cluster.sh
Starting cluster.
Starting standalonesession daemon on host ubuntu.
Starting taskexecutor daemon on host ubuntu.
[INFo] 1 instance(s) of taskexecutor are already running on ubuntu.
Starting taskexecutor daemon on host ubuntu.使用 Flink SQL 客户端工具连接集群导入相关依赖并开启 checkpoint
Flink SQL add jar /home/frankie/work/download/flink-sql-connector-mysql-cdc-2.4.0.jar;
[INFO] Execute statement succeed.
Flink SQL add jar /home/frankie/work/download/flink-sql-connector-pieclouddb-1.2.0.jar;
[INFO] Execute statement succeed.
Flink SQL SET execution.checkpointing.interval 3s;
[INFO] Execute statement succeed.创建 Flink CDC 源表
Flink SQL CREATE TABLE source_student_mysql ( id INT, name STRING, score INT, sex CHAR(1), PRIMARY KEY(id) NOT ENFORCED ) WITH ( connector mysql-cdc, hostname mysql-host, port 3306, username root, password 123456, database-name testdb, table-name student);
[INFO] Execute statement succeed.创建 Flink PieCloudDB 目标表
Flink SQL CREATE TABLE sink_student_pdb ( id INT, name STRING, score INT, sex CHAR(1), PRIMARY KEY(id) NOT ENFORCED ) WITH ( connector pieclouddb, hostname pieclouddb-host, port your-pieclouddb-port, username your-username, password your-password, pdb_warehouse your-pdbwarehouse, database-name demo, table-name student, load_mode merge);
[INFO] Execute statement succeed.执行导入
Flink SQL INSERT INTO sink_student_pdb SELECT * FROM source_student_mysql;
[INFO] Submitting SQL update statement to the cluster...
[INFO] SQL update statement has been successfully submitted to the cluster:
Job ID: 660b747ef8fb64f95064a461af9924bc 查看 Flink 的 WebUI可以看到这个数据导入的流任务在持续运行 Flink Web 操作界面
查看 PieCloudDB 中的数据可以看到数据已经正确导入 数据成功导入 PieCloudDB
除了使用 Flink SQL 的接入方式之外PieCloudDB Flink Connector 还支持通过 Flink Datastream API 来使用。
未来拓数派团队致力于对 PieCloudDB Flink Connector 进行持续的功能增强与迭代升级计划引入高级特性如 schema evolution 以及动态加表功能以满足更复杂的数据处理需求。
同时PieCloudDB 将持续扩展其数据同步工具组件的生态致力于打造更为全面和强大的连接工具包括 Flink、Spark 等大数据处理框架的集成工具以及 CDCChange Data Capture和 Kafka 等实时数据同步工具让用户将能够实现数据的高效流动和实时处理进一步释放数据的潜力。
