flink cdc多种数据源安装、配置与验证(超详细总结)(下)

本文涉及的产品
RDS SQL Server Serverless,2-4RCU 50GB 3个月
推荐场景:
实时计算 Flink 版,5000CU*H 3个月
云数据库 RDS SQL Server,基础系列 2核4GB
简介: flink cdc多种数据源安装、配置与验证(超详细总结)(下)

Step5:创建表空间

-- 以DBA的权限登录数据库
sqlplus /nolog
CONNECT sys/system AS SYSDBA
-- 创建一个名为"logminer_tbs"的表空间
-- 指定表空间的数据文件路径为"/home/oracle/app/oracle/product/11.2.0/logminer_tbs.dbf",其中"/home/oracle/app/oracle/product/11.2.0"是数据文件存储的目录,"logminer_tbs.dbf"是数据文件的文件名
-- 设置表空间的初始大小为25MB
-- 如果数据文件已经存在且可重用,将其重用,否则创建一个新的数据文件
-- 启用表空间的自动扩展功能,即当表空间空间不足时,自动增加数据文件的大小
-- 设置表空间的最大允许大小为无限,即表空间可以无限制地自动扩展
CREATE TABLESPACE logminer_tbs DATAFILE '/home/oracle/app/oracle/product/11.2.0/logminer_tbs.dbf' SIZE 25M REUSE AUTOEXTEND ON MAXSIZE UNLIMITED;

Step6:创建用户并赋权

-- 创建一个名为"flinkuser"的用户,密码为"flinkpw",将其默认表空间设置为"LOGMINER_TBS",并在该表空间上设置无限配额。
CREATE USER flinkuser IDENTIFIED BY flinkpw DEFAULT TABLESPACE LOGMINER_TBS QUOTA UNLIMITED ON LOGMINER_TBS;
-- 允许"flinkuser"用户创建会话,即允许该用户连接到数据库。
GRANT CREATE SESSION TO flinkuser;
-- (不支持Oracle 11g)允许"flinkuser"用户在多租户数据库(CDB)中设置容器。
-- GRANT SET CONTAINER TO flinkuser;
-- 允许"flinkuser"用户查询V_$DATABASE视图,该视图包含有关数据库实例的信息。
GRANT SELECT ON V_$DATABASE TO flinkuser;
-- 允许"flinkuser"用户执行任何表的闪回操作。
GRANT FLASHBACK ANY TABLE TO flinkuser;
-- 允许"flinkuser"用户查询任何表的数据。
GRANT SELECT ANY TABLE TO flinkuser;
-- 允许"flinkuser"用户拥有SELECT_CATALOG_ROLE角色,该角色允许查询数据字典和元数据。
GRANT SELECT_CATALOG_ROLE TO flinkuser;
-- 允许"flinkuser"用户拥有EXECUTE_CATALOG_ROLE角色,该角色允许执行一些数据字典中的过程和函数。
GRANT EXECUTE_CATALOG_ROLE TO flinkuser;
-- 允许"flinkuser"用户查询任何事务。
GRANT SELECT ANY TRANSACTION TO flinkuser;
-- (不支持Oracle 11g)允许"flinkuser"用户进行数据变更追踪(LogMiner)。
-- GRANT LOGMINING TO flinkuser;
-- 允许"flinkuser"用户创建表。
GRANT CREATE TABLE TO flinkuser;
-- 允许"flinkuser"用户锁定任何表。
GRANT LOCK ANY TABLE TO flinkuser;
-- 允许"flinkuser"用户修改任何表。
GRANT ALTER ANY TABLE TO flinkuser;
-- 允许"flinkuser"用户创建序列。
GRANT CREATE SEQUENCE TO flinkuser;
-- 允许"flinkuser"用户执行DBMS_LOGMNR包中的过程。
GRANT EXECUTE ON DBMS_LOGMNR TO flinkuser;
-- 允许"flinkuser"用户执行DBMS_LOGMNR_D包中的过程。
GRANT EXECUTE ON DBMS_LOGMNR_D TO flinkuser;
-- 允许"flinkuser"用户查询V_$LOG视图,该视图包含有关数据库日志文件的信息。
GRANT SELECT ON V_$LOG TO flinkuser;
-- 允许"flinkuser"用户查询V_$LOG_HISTORY视图,该视图包含有关数据库历史日志文件的信息。
GRANT SELECT ON V_$LOG_HISTORY TO flinkuser;
-- 允许"flinkuser"用户查询V_$LOGMNR_LOGS视图,该视图包含有关LogMiner日志文件的信息。
GRANT SELECT ON V_$LOGMNR_LOGS TO flinkuser;
-- 允许"flinkuser"用户查询V_$LOGMNR_CONTENTS视图,该视图包含LogMiner日志文件的内容。
GRANT SELECT ON V_$LOGMNR_CONTENTS TO flinkuser;
-- 允许"flinkuser"用户查询V_$LOGMNR_PARAMETERS视图,该视图包含有关LogMiner的参数信息。
GRANT SELECT ON V_$LOGMNR_PARAMETERS TO flinkuser;
-- 允许"flinkuser"用户查询V_$LOGFILE视图,该视图包含有关数据库日志文件的信息。
GRANT SELECT ON V_$LOGFILE TO flinkuser;
-- 允许"flinkuser"用户查询V_$ARCHIVED_LOG视图,该视图包含已归档的数据库日志文件的信息。
GRANT SELECT ON V_$ARCHIVED_LOG TO flinkuser;
-- 允许"flinkuser"用户查询V_$ARCHIVE_DEST_STATUS视图,该视图包含有关归档目标状态的信息。
GRANT SELECT ON V_$ARCHIVE_DEST_STATUS TO flinkuser;

Step7:数据库和表启用增量日志

-- 切换至flinkuser用户
sqlplus /nolog
CONNECT flinkuser/flinkpw
-- 创建customers表
CREATE TABLE customers (
    customer_id NUMBER PRIMARY KEY,
    customer_name VARCHAR2(50),
    email VARCHAR2(100),
    phone VARCHAR2(20)
) TABLESPACE LOGMINER_TBS;
-- 查看LOGMINER_TBS表空间下的所有表
select tablespace_name, table_name from user_tables
where tablespace_name = 'LOGMINER_TBS';
-- 以DBA的权限登录数据库
sqlplus /nolog
CONNECT sys/system AS SYSDBA
-- 为LOGMINER_TBS表空间下的customers表启用增强日志记录
ALTER TABLE FLINKUSER.CUSTOMERS ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS
-- 为数据库启用增强日志记录:
ALTER DATABASE ADD SUPPLEMENTAL LOG DATA;

2.4 SQLServer

版本:Microsoft SQL Server 2019 (RTM-CU21) (KB5025808) - 15.0.4316.3 (X64)

2.4.1 安装

Step1:拉取SQL Server 2019 镜像

docker pull mcr.microsoft.com/mssql/server:2019-latest

Step2:运行 SQL Server 容器(密码必须是8个字符,并包含字母、数字和特殊字符,如:abc@123456 ,下面映射主机端口为30027)

docker run -e 'ACCEPT_EULA=Y' -e 'SA_PASSWORD=abc@123456' -p 30027:1433 --name sql_server_2019 -d mcr.microsoft.com/mssql/server:2019-latest

Step3:验证 SQL Server 容器是否正在运行

docker ps -a|grep sql_server_2019

2.4.2 CDC 配置

Step1:开启SQLServer代理

## 使用root用户登录容器
docker exec -it --user root sql_server_2019 bash
## 进入容器后,执行命令启用Agent
/opt/mssql/bin/mssql-conf set sqlagent.enabled true
## 退出,重启容器
exit
docker restart sql_server_2019

Step2:创建’cdc_test’测试数据库,并使用连接工具登录该数据库,使用以下 SQL 命令启用 CDC 功能

-- 创建数据库
CREATE DATABASE cdc_test;
-- 启用CDC功能
EXEC sys.sp_cdc_enable_db;
-- 判断当前数据库是否启用了CDC(如果返回1,表示已启用)
SELECT is_cdc_enabled FROM sys.databases WHERE name = 'cdc_test';

Step3:选择要进行 CDC 跟踪的表(这里使用orders表作为演示

-- 创建示例表(orders)
CREATE TABLE orders (
     id int,
     order_date date,
     purchaser int,
     quantity int,
     product_id int,
     PRIMARY KEY ([id])
);
-- schema_name 是表所属的架构(schema)的名称。
-- table_name 是要启用 CDC 跟踪的表的名称。
-- cdc_role 是 CDC 使用的角色的名称。如果没有指定角色名称,系统将创建一个默认角色。
EXEC sys.sp_cdc_enable_table
  @source_schema = 'dbo',
  @source_name   = 'orders',
  @role_name     = 'cdc_role';

3. 验证

如果要验证flink cdc的功能,需要先下载flink的安装包,然后下载相应的cdc jar包并依赖,最后使用安装包里面的sql-client写相关的flink sql即可验证。

3.1 Flink版本与CDC版本的对应关系

下载Flink安装包以及jar包前,必须确定Flink CDC与Flink版本关系:

Flink CDC 版本 Flink 版本
1.0.0 1.11.*
1.1.0 1.11.*
1.2.0 1.12.*
1.3.0 1.12.*
1.4.0 1.13.*
2.0.* 1.13.*
2.1.* 1.13.*
2.2.* 1.13.*, 1.14.*
2.3.* 1.13.*, 1.14.*, 1.15.*, 1.16.0
2.4.* 1.13.*, 1.14.*, 1.15.*, 1.16.*, 1.17.0

本文以 Flink1.13.6 + Flink CDC 2.2.0 版本为例子演示。

3.2 下载相关包

flink 安装包下载,下载地址:https://flink.apache.org/downloads/

下载cdc相关的jar,根据自己的需求,下载相关的cdc jar:https://repo1.maven.org/maven2/com/ververica/

3.3 添加cdc jar 至lib目录

把需要验证的cdc jar放到flink安装包解压之后的lib目录(<FLINK_HOME>/lib/):

3.4 验证

使用下面的命令启动 Flink 集群:

./bin/start-cluster.sh

启动成功,可以访问 http://localhost:8081 访问到 Flink Web UI:

使用下面的命令启动 Flink SQL CLI :

./bin/sql-client.sh

展示如下页面,表示启动flink客户端成功:

执行如下FlinkSQL:

CREATE TABLE t_source_sqlserver (
   id INT,
    order_date DATE,
    purchaser INT,
    quantity INT,
    product_id INT,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'sqlserver-cdc',
  'hostname' = '10.194.183.120',
  'port' = '30027',
  'username' = 'sa',
  'password' = 'abc@123456',
  'database-name' = 'cdc_test',
  'schema-name' = 'dbo',
  'table-name' = 'orders'
);

可以看到执行成功了:

执行select 语句,以便实时查看该表的数据变动:

select * from t_source_sqlserver;

从下图,可以看出,只要修改左边的数据,会在控制台实时显示新增删除的数据。

同时,也能在Flink web页面看到任务正在运行:

最后,可以通过如下命令关闭掉Flink启动的集群:

./stop-cluster.sh
相关实践学习
基于Hologres轻松玩转一站式实时仓库
本场景介绍如何利用阿里云MaxCompute、实时计算Flink和交互式分析服务Hologres开发离线、实时数据融合分析的数据大屏应用。
Linux入门到精通
本套课程是从入门开始的Linux学习课程,适合初学者阅读。由浅入深案例丰富,通俗易懂。主要涉及基础的系统操作以及工作中常用的各种服务软件的应用、部署和优化。即使是零基础的学员,只要能够坚持把所有章节都学完,也一定会受益匪浅。
目录
相关文章
|
4月前
|
分布式计算 资源调度 大数据
大数据-110 Flink 安装部署 下载解压配置 Standalone模式启动 打包依赖(一)
大数据-110 Flink 安装部署 下载解压配置 Standalone模式启动 打包依赖(一)
129 0
|
17天前
|
消息中间件 关系型数据库 MySQL
Flink CDC 在阿里云实时计算Flink版的云上实践
本文整理自阿里云高级开发工程师阮航在Flink Forward Asia 2024的分享,重点介绍了Flink CDC与实时计算Flink的集成、CDC YAML的核心功能及应用场景。主要内容包括:Flink CDC的发展及其在流批数据处理中的作用;CDC YAML支持的同步链路、Transform和Route功能、丰富的监控指标;典型应用场景如整库同步、Binlog原始数据同步、分库分表同步等;并通过两个Demo展示了MySQL整库同步到Paimon和Binlog同步到Kafka的过程。最后,介绍了未来规划,如脏数据处理、数据限流及扩展数据源支持。
154 0
Flink CDC 在阿里云实时计算Flink版的云上实践
|
2月前
|
监控 关系型数据库 MySQL
Flink CDC MySQL同步MySQL错误记录
在使用Flink CDC同步MySQL数据时,常见的错误包括连接错误、权限错误、表结构变化、数据类型不匹配、主键冲突和
146 16
|
3月前
|
消息中间件 资源调度 关系型数据库
如何在Flink on YARN环境中配置Debezium CDC 3.0,以实现实时捕获数据库变更事件并将其传输到Flink进行处理
本文介绍了如何在Flink on YARN环境中配置Debezium CDC 3.0,以实现实时捕获数据库变更事件并将其传输到Flink进行处理。主要内容包括安装Debezium、配置Kafka Connect、创建Flink任务以及启动任务的具体步骤,为构建实时数据管道提供了详细指导。
186 9
|
4月前
|
消息中间件 NoSQL Kafka
大数据-116 - Flink DataStream Sink 原理、概念、常见Sink类型 配置与使用 附带案例1:消费Kafka写到Redis
大数据-116 - Flink DataStream Sink 原理、概念、常见Sink类型 配置与使用 附带案例1:消费Kafka写到Redis
271 0
|
5月前
|
运维 数据处理 数据安全/隐私保护
阿里云实时计算Flink版测评报告
该测评报告详细介绍了阿里云实时计算Flink版在用户行为分析与标签画像中的应用实践,展示了其毫秒级的数据处理能力和高效的开发流程。报告还全面评测了该服务在稳定性、性能、开发运维及安全性方面的卓越表现,并对比自建Flink集群的优势。最后,报告评估了其成本效益,强调了其灵活扩展性和高投资回报率,适合各类实时数据处理需求。
|
3月前
|
存储 分布式计算 流计算
实时计算 Flash – 兼容 Flink 的新一代向量化流计算引擎
本文介绍了阿里云开源大数据团队在实时计算领域的最新成果——向量化流计算引擎Flash。文章主要内容包括:Apache Flink 成为业界流计算标准、Flash 核心技术解读、性能测试数据以及在阿里巴巴集团的落地效果。Flash 是一款完全兼容 Apache Flink 的新一代流计算引擎,通过向量化技术和 C++ 实现,大幅提升了性能和成本效益。
1619 73
实时计算 Flash – 兼容 Flink 的新一代向量化流计算引擎
zdl
|
3月前
|
消息中间件 运维 大数据
大数据实时计算产品的对比测评:实时计算Flink版 VS 自建Flink集群
本文介绍了实时计算Flink版与自建Flink集群的对比,涵盖部署成本、性能表现、易用性和企业级能力等方面。实时计算Flink版作为全托管服务,显著降低了运维成本,提供了强大的集成能力和弹性扩展,特别适合中小型团队和业务波动大的场景。文中还提出了改进建议,并探讨了与其他产品的联动可能性。总结指出,实时计算Flink版在简化运维、降低成本和提升易用性方面表现出色,是大数据实时计算的优选方案。
zdl
195 56
|
2月前
|
存储 关系型数据库 BI
实时计算UniFlow:Flink+Paimon构建流批一体实时湖仓
实时计算架构中,传统湖仓架构在数据流量管控和应用场景支持上表现良好,但在实际运营中常忽略细节,导致新问题。为解决这些问题,提出了流批一体的实时计算湖仓架构——UniFlow。该架构通过统一的流批计算引擎、存储格式(如Paimon)和Flink CDC工具,简化开发流程,降低成本,并确保数据一致性和实时性。UniFlow还引入了Flink Materialized Table,实现了声明式ETL,优化了调度和执行模式,使用户能灵活调整新鲜度与成本。最终,UniFlow不仅提高了开发和运维效率,还提供了更实时的数据支持,满足业务决策需求。
|
6月前
|
SQL 消息中间件 Kafka
实时计算 Flink版产品使用问题之如何在EMR-Flink的Flink SOL中针对source表单独设置并行度
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。