PieCloudDB Database Flink Connector:让数据流动起来

2024-06-01 10:44

本文主要是介绍PieCloudDB Database Flink Connector:让数据流动起来,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

面对客户环境中长期运行的各种类型的传统数据库,如何优雅地设计数据迁移的方案,既能灵活地应对各种数据导入场景和多源异构数据库,又能满足客户对数据导入结果的准确性、一致性、实时性的要求,让客户平滑地迁移到 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 机制和幂等写过程如下图所示:

image.png

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.student"Column | 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,可以看到这个数据导入的流任务在持续运行:

image.png

Flink Web 操作界面

查看 PieCloudDB 中的数据,可以看到数据已经正确导入

image.png

数据成功导入 PieCloudDB

除了使用 Flink SQL 的接入方式之外,PieCloudDB Flink Connector 还支持通过 Flink Datastream API 来使用。

未来,拓数派团队致力于对 PieCloudDB Flink Connector 进行持续的功能增强与迭代升级,计划引入高级特性,如 schema evolution 以及动态加表功能,以满足更复杂的数据处理需求。

同时,PieCloudDB 将持续扩展其数据同步工具组件的生态,致力于打造更为全面和强大的连接工具,包括 Flink、Spark 等大数据处理框架的集成工具,以及 CDC(Change Data Capture)和 Kafka 等实时数据同步工具,让用户将能够实现数据的高效流动和实时处理,进一步释放数据的潜力。

这篇关于PieCloudDB Database Flink Connector:让数据流动起来的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1020785

相关文章

大模型研发全揭秘:客服工单数据标注的完整攻略

在人工智能(AI)领域,数据标注是模型训练过程中至关重要的一步。无论你是新手还是有经验的从业者,掌握数据标注的技术细节和常见问题的解决方案都能为你的AI项目增添不少价值。在电信运营商的客服系统中,工单数据是客户问题和解决方案的重要记录。通过对这些工单数据进行有效标注,不仅能够帮助提升客服自动化系统的智能化水平,还能优化客户服务流程,提高客户满意度。本文将详细介绍如何在电信运营商客服工单的背景下进行

SQL中的外键约束

外键约束用于表示两张表中的指标连接关系。外键约束的作用主要有以下三点: 1.确保子表中的某个字段(外键)只能引用父表中的有效记录2.主表中的列被删除时,子表中的关联列也会被删除3.主表中的列更新时,子表中的关联元素也会被更新 子表中的元素指向主表 以下是一个外键约束的实例展示

基于MySQL Binlog的Elasticsearch数据同步实践

一、为什么要做 随着马蜂窝的逐渐发展,我们的业务数据越来越多,单纯使用 MySQL 已经不能满足我们的数据查询需求,例如对于商品、订单等数据的多维度检索。 使用 Elasticsearch 存储业务数据可以很好的解决我们业务中的搜索需求。而数据进行异构存储后,随之而来的就是数据同步的问题。 二、现有方法及问题 对于数据同步,我们目前的解决方案是建立数据中间表。把需要检索的业务数据,统一放到一张M

关于数据埋点,你需要了解这些基本知识

产品汪每天都在和数据打交道,你知道数据来自哪里吗? 移动app端内的用户行为数据大多来自埋点,了解一些埋点知识,能和数据分析师、技术侃大山,参与到前期的数据采集,更重要是让最终的埋点数据能为我所用,否则可怜巴巴等上几个月是常有的事。   埋点类型 根据埋点方式,可以区分为: 手动埋点半自动埋点全自动埋点 秉承“任何事物都有两面性”的道理:自动程度高的,能解决通用统计,便于统一化管理,但个性化定

如何去写一手好SQL

MySQL性能 最大数据量 抛开数据量和并发数,谈性能都是耍流氓。MySQL没有限制单表最大记录数,它取决于操作系统对文件大小的限制。 《阿里巴巴Java开发手册》提出单表行数超过500万行或者单表容量超过2GB,才推荐分库分表。性能由综合因素决定,抛开业务复杂度,影响程度依次是硬件配置、MySQL配置、数据表设计、索引优化。500万这个值仅供参考,并非铁律。 博主曾经操作过超过4亿行数据

使用SecondaryNameNode恢复NameNode的数据

1)需求: NameNode进程挂了并且存储的数据也丢失了,如何恢复NameNode 此种方式恢复的数据可能存在小部分数据的丢失。 2)故障模拟 (1)kill -9 NameNode进程 [lytfly@hadoop102 current]$ kill -9 19886 (2)删除NameNode存储的数据(/opt/module/hadoop-3.1.4/data/tmp/dfs/na

异构存储(冷热数据分离)

异构存储主要解决不同的数据,存储在不同类型的硬盘中,达到最佳性能的问题。 异构存储Shell操作 (1)查看当前有哪些存储策略可以用 [lytfly@hadoop102 hadoop-3.1.4]$ hdfs storagepolicies -listPolicies (2)为指定路径(数据存储目录)设置指定的存储策略 hdfs storagepolicies -setStoragePo

Hadoop集群数据均衡之磁盘间数据均衡

生产环境,由于硬盘空间不足,往往需要增加一块硬盘。刚加载的硬盘没有数据时,可以执行磁盘数据均衡命令。(Hadoop3.x新特性) plan后面带的节点的名字必须是已经存在的,并且是需要均衡的节点。 如果节点不存在,会报如下错误: 如果节点只有一个硬盘的话,不会创建均衡计划: (1)生成均衡计划 hdfs diskbalancer -plan hadoop102 (2)执行均衡计划 hd

性能分析之MySQL索引实战案例

文章目录 一、前言二、准备三、MySQL索引优化四、MySQL 索引知识回顾五、总结 一、前言 在上一讲性能工具之 JProfiler 简单登录案例分析实战中已经发现SQL没有建立索引问题,本文将一起从代码层去分析为什么没有建立索引? 开源ERP项目地址:https://gitee.com/jishenghua/JSH_ERP 二、准备 打开IDEA找到登录请求资源路径位置

MySQL数据库宕机,启动不起来,教你一招搞定!

作者介绍:老苏,10余年DBA工作运维经验,擅长Oracle、MySQL、PG、Mongodb数据库运维(如安装迁移,性能优化、故障应急处理等)公众号:老苏畅谈运维欢迎关注本人公众号,更多精彩与您分享。 MySQL数据库宕机,数据页损坏问题,启动不起来,该如何排查和解决,本文将为你说明具体的排查过程。 查看MySQL error日志 查看 MySQL error日志,排查哪个表(表空间