Flink connector Oracle CDC synchronizes data to MySQL in real time (oracle19c)

preparation

In this step, you need to configure Oracle. It mainly includes .

1. Turn on Archive log
2. Open the database and data table supplemental log
3. establish CDC Users and permissions

Be careful ： Do not use Oracle Of SYS and SYSTEM User as CDC user . Because these two users can capture a large number of Oracle Change information inside the database , It is unnecessary for business data .Debezium The changes captured by these two users will be filtered out .

Let's start the configuration steps . In the installation Oracle On the machine ：

su - oracle
sqlplus / as sysdba

Get into Sqlplus. Then open Archive log.

alter system set db_recovery_file_dest_size = 10G;
alter system set db_recovery_file_dest = '/opt/oracle/oradata/recovery_area' scope=spfile;
shutdown immediate;
startup mount;
alter database archivelog;
alter database open;
#  Check Archive log Is it successfully opened 
archive log list;

Be careful ：'/opt/oracle/oradata/recovery_area' If this path does not exist, you need to create it manually （Oracle Create... Under user ）

1. This step requires restarting the database , Please choose to operate at the right time .
2. In the example /opt/oracle/oradata/recovery_area Catalog oracle Users need to have read and write permission .
3. If you execute alter system set db_recovery_file_dest = '/opt/oracle/oradata/recovery_area' scope=spfile; Time report ORA-32001: write to SPFILE requested but no SPFILE is in use. Need to check spfile file .

show parameter spfile;
#  If the output value It's empty , Description... Was not created spfile, Do the following SQL establish 
create spfile from pfile;
#  Shut down and restart 
shutdown immediate;
startup;
#  Check spfile Successful creation of 
show parameter spfile;

Next, make relevant important configuration :

--  Check whether the log archive is enabled 
archive log list;

--  Enable supplementary logging for the captured database , So that the data changes capture the state before the changed database rows , The following describes how to configure at the database level .
ALTER DATABASE ADD SUPPLEMENTAL LOG DATA;

-  Create tablespace 
CREATE TABLESPACE logminer_tbs DATAFILE '/home/oracle/logminer_tbs.dbf' SIZE 25M REUSE AUTOEXTEND ON MAXSIZE UNLIMITED;

--  Create user family Bound tablespace LOGMINER_TBS
CREATE USER C##family IDENTIFIED BY zyhcdc DEFAULT TABLESPACE LOGMINER_TBS QUOTA UNLIMITED ON LOGMINER_TBS;

--  grant family user dba Authority 
 grant connect,resource,dba to C##family;

-  And grant permission 
GRANT CREATE SESSION TO C##family;
GRANT SELECT ON V_$DATABASE to C##family;
GRANT FLASHBACK ANY TABLE TO C##family;
GRANT SELECT ANY TABLE TO C##family;
GRANT SELECT_CATALOG_ROLE TO C##family;
GRANT EXECUTE_CATALOG_ROLE TO C##family;
GRANT SELECT ANY TRANSACTION TO C##family;
GRANT EXECUTE ON SYS.DBMS_LOGMNR TO C##family;
GRANT SELECT ON V_$LOGMNR_CONTENTS TO C##family;
GRANT CREATE TABLE TO C##family;
GRANT LOCK ANY TABLE TO C##family;
GRANT ALTER ANY TABLE TO C##family;
GRANT CREATE SEQUENCE TO C##family;

GRANT EXECUTE ON DBMS_LOGMNR TO C##family;
GRANT EXECUTE ON DBMS_LOGMNR_D TO C##family;

GRANT SELECT ON V_$LOG TO C##family;
GRANT SELECT ON V_$LOG_HISTORY TO C##family;
GRANT SELECT ON V_$LOGMNR_LOGS TO C##family;
GRANT SELECT ON V_$LOGMNR_CONTENTS TO C##family;
GRANT SELECT ON V_$LOGMNR_PARAMETERS TO C##family;
GRANT SELECT ON V_$LOGFILE TO C##family;
GRANT SELECT ON V_$ARCHIVED_LOG TO C##family;
GRANT SELECT ON V_$ARCHIVE_DEST_STATUS TO C##family;

For local use Navicate Connect Oracle：

--  establish  STUDENT_INFO  surface 
create table student_info (
  sid         number(10) constraint pk_sid primary key,
  sname       varchar2(10),
  sex         varchar2(2)
);

--  modify STUDENT_INFO Table to support incremental logging , First in Oracle Created in STUDENT_INFO The table executes the following statement 
ALTER TABLE C##FAMILY.STUDENT_INFO ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;

thus ,Oracle Configuration complete ;

Flink CDC To configure ：

Here we use Flink SQL CLI Do a demonstration ：

Use the following command to switch to Flink Catalog

cd flink-1.14.0

  Start with the following command Flink SQL CLI：

./bin/sql-client.sh

We should see CLI Welcome screen of the client .

stay Flink SQL CLI Use in Flink DDL Create table

Create a table that captures changed data from the corresponding database table .

Oracle CDC Tables can be defined as follows ：

-- register an Oracle table 'student_info' in Flink SQL
Flink SQL> CREATE TABLE student_info (
     SID INT NOT NULL,
     SNAME STRING,
     SEX STRING,
PRIMARY KEY(SID) NOT ENFORCED
     ) WITH (
     'connector' = 'oracle-cdc',
     'hostname' = 'localhost',
     'port' = '1521',
     'username' = 'C##family',
     'password' = 'zyhcdc',
     'database-name' = 'ORCLCDB',
     'schema-name' = 'C##FAMILY',
     'table-name' = 'STUDENT_INFO');
  
-- read snapshot and binlogs from products table
Flink SQL> SELECT * FROM student_info;

In defining a data outflow to MySQL Table of ：

-- register an Oracle table 'mysql_user' in Flink SQL
Flink SQL> CREATE TABLE mysql_user (
     SID INT ,
     SNAME STRING,
     SEX STRING,
PRIMARY KEY(SID) NOT ENFORCED
    ) WITH (
     'connector' = 'jdbc',
     'url' = 'jdbc:mysql://localhost:3306/test_cdc',
     'username' = 'root',
     'password' = 'root123',
     'table-name' = 'user'
    );

Flink SQL> insert into mysql_user select SID,SNAME,SEX from student_info;
[INFO] Submitting SQL update statement to the cluster...
[INFO] SQL update statement has been successfully submitted to the cluster:
Job ID: 1966fdd63bb36c14908fe8e31408db58

Complete the above configuration ;Flink connector Oracle CDC Real time data synchronization to MySQL That's it ; You can test it yourself later .

If there is a primary key, it will be automatically merge.