背景
flink cdc抽取cdb rac模式的oracle集群，抽取一直存在賬號(hào)權(quán)限問(wèn)題
前提條件
開(kāi)啟歸檔日志
rac 模式歸檔日志需要在共享存儲(chǔ)中薇缅。如果配置在本地，會(huì)導(dǎo)致歸檔日志找不到
一懦鼠、開(kāi)啟歸檔日志
位置要指定為共享存儲(chǔ)容诬，位置可以自行調(diào)整
alter system set db_recovery_file_dest='+DATA' scope=spfile sid='*';
alter system set db_recovery_file_dest_size=2048M scope=spfile sid='*';
重啟數(shù)據(jù)庫(kù)實(shí)例
備注： rac19cdb是數(shù)據(jù)庫(kù)實(shí)例名稱(chēng)，rac19cdb1數(shù)據(jù)庫(kù)節(jié)點(diǎn)名稱(chēng)秩彤，需自行調(diào)整叔扼，下同
su oracle
srvctl stop database -d rac19cdb
srvctl start instance -d rac19cdb -i rac19cdb1 -o mount
執(zhí)行sql,開(kāi)啟歸檔
alter database archivelog;
啟動(dòng)另外一臺(tái)數(shù)據(jù)庫(kù)實(shí)例
srvctl start instance -d rac19cdb -i rac19cdb2 -o mount
二、 在cdb賬號(hào)下創(chuàng)建cdb表空間
create tablespace logminer_tbs datafile '+DATA' size 100M reuse autoextend on next 100M maxsize unlimited
備注： +DATA是共享存儲(chǔ)位置漫雷，大小瓜富，可自行調(diào)整
三、 切換到pdb
alter session set container=RACDBPDB
備注： RACDBPDB是pdb名稱(chēng)降盹，需自行調(diào)整
四与柑、 在pdb創(chuàng)建表空間
create tablespace logminer_tbs datafile '+DATA' size 100M reuse autoextend on next 100M maxsize unlimited
備注： +DATA是共享存儲(chǔ)位置，可自行調(diào)整
五、在cdb下創(chuàng)建cdb用戶(hù)并授權(quán)
CREATE USER c##flinkuser IDENTIFIED BY flinkpw DEFAULT TABLESPACE logminer_tbs QUOTA UNLIMITED ON logminer_tbs CONTAINER=ALL;
GRANT CREATE SESSION TO c##flinkuser CONTAINER=ALL;
GRANT SET CONTAINER TO c##flinkuser CONTAINER=ALL;
GRANT SELECT ON V_$DATABASE to c##flinkuser CONTAINER=ALL;
GRANT FLASHBACK ANY TABLE TO c##flinkuser CONTAINER=ALL;
GRANT SELECT ANY TABLE TO c##flinkuser CONTAINER=ALL;
GRANT SELECT_CATALOG_ROLE TO c##flinkuser CONTAINER=ALL;
GRANT EXECUTE_CATALOG_ROLE TO c##flinkuser CONTAINER=ALL;
GRANT SELECT ANY TRANSACTION TO c##flinkuser CONTAINER=ALL;
GRANT LOGMINING TO c##flinkuser CONTAINER=ALL;
GRANT CREATE TABLE TO c##flinkuser CONTAINER=ALL;
GRANT LOCK ANY TABLE TO c##flinkuser CONTAINER=ALL;
GRANT CREATE SEQUENCE TO c##flinkuser CONTAINER=ALL;
GRANT EXECUTE ON DBMS_LOGMNR TO c##flinkuser CONTAINER=ALL;
GRANT EXECUTE ON DBMS_LOGMNR_D TO c##flinkuser CONTAINER=ALL;
GRANT SELECT ON V_$LOG TO c##flinkuser CONTAINER=ALL;
GRANT SELECT ON V_$LOG_HISTORY TO c##flinkuser CONTAINER=ALL;
GRANT SELECT ON V_$LOGMNR_LOGS TO c##flinkuser CONTAINER=ALL;
GRANT SELECT ON V_$LOGMNR_CONTENTS TO c##flinkuser CONTAINER=ALL;
GRANT SELECT ON V_$LOGMNR_PARAMETERS TO c##flinkuser CONTAINER=ALL;
GRANT SELECT ON V_$LOGFILE TO c##flinkuser CONTAINER=ALL;
GRANT SELECT ON V_$ARCHIVED_LOG TO c##flinkuser CONTAINER=ALL;
GRANT SELECT ON V_$ARCHIVE_DEST_STATUS TO c##flinkuser CONTAINER=ALL;
備注：授權(quán)語(yǔ)句需要全量執(zhí)行
六价捧、 在pdb下創(chuàng)建pdb用戶(hù)
create USER test IDENTIFIED BY test123 DEFAULT TABLESPACE logminer_tbs QUOTA UNLIMITED ON logminer_tbs
GRANT CREATE SESSION TO test;
GRANT CREATE TABLE TO test;
GRANT LOCK ANY TABLE TO test;
GRANT ALTER ANY TABLE TO test;
GRANT CREATE SEQUENCE TO test;
七丑念、 在pdb賬號(hào)下測(cè)試數(shù)據(jù)準(zhǔn)備
1、 創(chuàng)建表
CREATE TABLE TEST."gx_fen" (
"id" NUMBER(11,0),
"name" VARCHAR2(255),
"price" VARCHAR2(20),
"description" VARCHAR2(255),
"inc_day" VARCHAR2(255),
"ts" TIMESTAMP,
"dt" TIMESTAMP WITH LOCAL TIME ZONE,
CONSTRAINT SYS_C007562 CHECK ("id" IS NOT NULL)
);
2结蟋、 表開(kāi)啟附加日志
ALTER TABLE TEST."gx_fen" ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
3脯倚、插入數(shù)據(jù)
INSERT INTO TEST."gx_fen" ("id", "name", "price", "description", "inc_day", "ts", "dt") VALUES(1, '螺', '10.00', 'luo si fen', '2022-05-25', TIMESTAMP '2022-05-25 13:07:25.000000', TIMESTAMP '2022-05-25 05:07:25.000000');
INSERT INTO TEST."gx_fen" ("id", "name", "price", "description", "inc_day", "ts", "dt") VALUES(2, '桂', '9.00', 'gui lin mi fen', '2022-05-25', TIMESTAMP '2022-05-25 13:07:37.000000', TIMESTAMP '2022-05-25 05:07:37.000000');
4、創(chuàng)建finksql任務(wù)并啟動(dòng)椎眯，觀察日志挠将，抽取到全量數(shù)據(jù)則為正常
CREATE TABLE products (
id INT,
name STRING
) WITH (
'connector' = 'oracle-cdc',
'hostname' = '1111111111',
'port' = '111111111',
'username' = 'c##flinkuser', #cdb用戶(hù)名
'password' = 'flinkpw', #cdb用戶(hù)密碼
'database-name' = 'racdb', #racdb 名稱(chēng)
'schema-name' = 'test', #pdb用戶(hù)名
'table-name' = 'gx_fen', #表名
'debezium.database.pdb.name' = 'racdbpdb'，###showpdbs 查看到名
'debezium.log.mining.strategy' = 'online_catalog'
)
5编整、在pdb賬號(hào)下再次插入數(shù)據(jù)舔稀，等待日志更新，抽取到新增數(shù)據(jù)則為正常
INSERT INTO TEST."gx_fen" ("id", "name", "price", "description", "inc_day", "ts", "dt") VALUES(4, '老友', '13', '南寧', '2023-01-09', TIMESTAMP '2023-01-09 15:10:39.000000', TIMESTAMP '2023-01-09 07:10:39.000000');
INSERT INTO TEST."gx_fen" ("id", "name", "price", "description", "inc_day", "ts", "dt") VALUES(6, '老友1', '13', '老友', '2023-01-09', TIMESTAMP '2023-01-09 16:16:39.000000', TIMESTAMP '2023-01-09 09:16:39.000000');
INSERT INTO TEST."gx_fen" ("id", "name", "price", "description", "inc_day", "ts", "dt") VALUES(7, '老友2', '10', '老友', '2023-01-10', TIMESTAMP '2023-01-10 16:16:39.000000', TIMESTAMP '2023-01-10 08:16:39.000000');
