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