Confluent Platform: ksqlDB 实时数据 ETL2 (Oracle-cdc Source)

亦凉 2023-01-23 05:56 39阅读 0赞

### 文章目录 ###

*  1, oracle(11g) cdc 配置
 *  2, 配置 oraclecdc-source
 *   *  2.1, redo log对应的topic: 包含执行的sql语句
     *  2.2, record event 对应的topic: 一条表数据
 *  3, oraclecdc-- > hbase

# 1, oracle(11g) cdc 配置 #

oraclecdc-source 配置文档：[https://docs.confluent.io/kafka-connect-oracle-cdc/current/overview.html][https_docs.confluent.io_kafka-connect-oracle-cdc_current_overview.html]

#1, Oracle 开启archeve/supplemental log
    SQL> SHUTDOWN IMMEDIATE;
     STARTUP MOUNT;
     ALTER DATABASE ARCHIVELOG;
     ALTER DATABASE OPEN;
     #To enable full supplemental logging for specific tables:
     ALTER DATABASE ADD SUPPLEMENTAL LOG DATA;
     ALTER TABLE <schema name>.<table name> ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
     -- ALTER DATABASE ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
    
    SQL> select name,log_mode from v$database;
    NAME      LOG_MODE
    --------- ------------
    HELOWIN   ARCHIVELOG
    	
    /* 防止归档日志激增： RMAN> connect target system/pass RMAN> crosscheck archivelog all; #删除过期的归档 RMAN> configure retention policy to recovery window of 3 days; --查看归档日志保留策略 RMAN> show archivelog deletion policy; -- 默认NONE，不删除 --修改归档日志保留策略为2个备份副本 RMAN> configure archivelog deletion policy to backed up 2 times to device type disk; RMAN> delete noprompt expired archivelog all; SQL> show parameter db_recovery; #增大闪回恢复区的大小 SQL>alter system set DB_RECOVERY_FILE_DEST_SIZE=8G; #查看ARCHIVELOG日志使用率 SQL> SELECT * FROM V$FLASH_RECOVERY_AREA_USAGE; */
    #2, 创建测试表
    SQL> alter user scott account unlock;
     alter user scott identified by scott;
     conn scott/scott
     create table test ( id int , name varchar2(10), mark varchar2(10));
    
    #3, 创建cdc 用户和权限
    conn / as sysdba
    CREATE ROLE CDC_PRIVS;
    GRANT CREATE SESSION TO CDC_PRIVS;
    GRANT EXECUTE ON SYS.DBMS_LOGMNR TO CDC_PRIVS;
    GRANT SELECT ON V_$LOGMNR_CONTENTS TO CDC_PRIVS;
    GRANT SELECT ON V_$DATABASE TO CDC_PRIVS;
    GRANT SELECT ON V_$THREAD TO CDC_PRIVS;
    GRANT SELECT ON V_$PARAMETER TO CDC_PRIVS;
    GRANT SELECT ON V_$NLS_PARAMETERS TO CDC_PRIVS;
    GRANT SELECT ON V_$TIMEZONE_NAMES TO CDC_PRIVS;
    GRANT SELECT ON ALL_INDEXES TO CDC_PRIVS;
    GRANT SELECT ON ALL_OBJECTS TO CDC_PRIVS;
    GRANT SELECT ON ALL_USERS TO CDC_PRIVS;
    GRANT SELECT ON ALL_CATALOG TO CDC_PRIVS;
    GRANT SELECT ON ALL_CONSTRAINTS TO CDC_PRIVS;
    GRANT SELECT ON ALL_CONS_COLUMNS TO CDC_PRIVS;
    GRANT SELECT ON ALL_TAB_COLS TO CDC_PRIVS;
    GRANT SELECT ON ALL_IND_COLUMNS TO CDC_PRIVS;
    GRANT SELECT ON ALL_ENCRYPTED_COLUMNS TO CDC_PRIVS;
    GRANT SELECT ON ALL_LOG_GROUPS TO CDC_PRIVS;
    GRANT SELECT ON ALL_TAB_PARTITIONS TO CDC_PRIVS;
    GRANT SELECT ON SYS.DBA_REGISTRY TO CDC_PRIVS;
    GRANT SELECT ON SYS.OBJ$ TO CDC_PRIVS;
    GRANT SELECT ON DBA_TABLESPACES TO CDC_PRIVS;
    GRANT SELECT ON DBA_OBJECTS TO CDC_PRIVS;
    GRANT SELECT ON SYS.ENC$ TO CDC_PRIVS;
    GRANT SELECT ANY TRANSACTION TO CDC_PRIVS;
    GRANT SELECT any table to CDC_PRIVS;
    --GRANT SELECT ON scott.test TO CDC_PRIVS;
    
    CREATE USER CDC_USER IDENTIFIED BY CDC_USER DEFAULT TABLESPACE USERS;
    ALTER USER CDC_USER QUOTA UNLIMITED ON USERS;
    GRANT CDC_PRIVS to CDC_USER;
    GRANT FLASHBACK ANY TABLE TO CDC_USER;
    
    #4, 检查用户权限
    SQL> conn / as sysdba
    SQL> set lines 256
    SQL> set pages 42
    SELECT GRANTEE, OWNER, TABLE_NAME FROM DBA_TAB_PRIVS
          WHERE GRANTEE IN (SELECT granted_role
                      FROM DBA_ROLE_PRIVS
                      WHERE GRANTEE = 'CDC_USER')
    AND (TABLE_NAME='DBMS_LOGMNR' OR TABLE_NAME='V_$LOGMNR_CONTENTS');  
    
    GRANTEE                        OWNER                          TABLE_NAME
    ------------------------------ ------------------------------ ------------------------------
    CDC_PRIVS                      SYS                            V_$LOGMNR_CONTENTS
    CDC_PRIVS                      SYS                            DBMS_LOGMNR

# 2, 配置 oraclecdc-source #

注意：

*  table.inclusion.regex: ORCL.SCOTT.TEST (实例名.用户名.表名)
 *  key.template: 默认 $\{primaryKeyStructOrValue\}，即使用primary key (无主键可以自定义key使用的字段）

SQL> select instance_name from  V$instance;
    INSTANCE_NAME
    ----------------
    helowin
    
    SQL> conn scott/scott;
    Connected.
    SQL> show user;
    USER is "SCOTT"
    
    SQL> select * from tab;
    TNAME                          TABTYPE  CLUSTERID
    ------------------------------ ------- ----------
    DEPT                           TABLE
    EMP                            TABLE
    TEST                           TABLE
    TEST_VIEW                      VIEW
    4 rows selected.
    
    #插入一条测试数据
    SQL> insert into test values(1,'a','a');

创建source connector (根据sid.username.tablename ,来获取源数据)：

ksql> CREATE SOURCE CONNECTOR `SimpleOracleCDC_25` WITH( 
       "connector.class"='io.confluent.connect.oracle.cdc.OracleCdcSourceConnector',
       "name"='SimpleOracleCDC_25',
       "tasks.max"=3,
       "key.converter"='org.apache.kafka.connect.storage.StringConverter',
       "key.template"='ID,NAME',
     
       "value.converter"='io.confluent.connect.avro.AvroConverter',
       "value.converter.schema.registry.url"='http://localhost:8081',
       "value.converter.schemas.enable"='false',
       "confluent.topic.bootstrap.servers"='localhost:9092',
    
        "oracle.server"='192.168.56.117',
        "oracle.port"='1523',
        "oracle.sid"='helowin',
        "oracle.username"='system',
        "oracle.password"='system',
       "start.from"='snapshot',
    
       "redo.log.topic.name"='SimpleOracleCDC_25',
       "redo.log.consumer.bootstrap.servers"='localhost:9092',
       "table.inclusion.regex"='helowin.SCOTT.TEST',
       "numeric.mapping"='best_fit',
       "_table.topic.name.template_"='Using template vars to set change event topic for each table',
       "table.topic.name.template"='${databaseName}25.${schemaName}25.${tableName}25',
       "connection.pool.max.size"= 20,
       "confluent.topic.replication.factor"=1,
       "topic.creation.groups"='redo',
       "topic.creation.redo.include"='SimpleOracleCDC_25',
       "topic.creation.redo.replication.factor"= 1,
       "topic.creation.redo.partitions"= 1,
       "topic.creation.redo.cleanup.policy"='delete',
       "topic.creation.redo.retention.ms"= 1209600000,
       "topic.creation.default.replication.factor"= 1,
       "topic.creation.default.partitions"= 1,
       "topic.creation.default.cleanup.policy"='compact'
    );
    
    shell> curl -s -X POST -H 'Content-Type: application/json'  http://localhost:8083/connectors  -d ' { "name": "SimpleOracleCDC_24", "config":{ "connector.class": "io.confluent.connect.oracle.cdc.OracleCdcSourceConnector", "name": "SimpleOracleCDC_24", "tasks.max":3, "key.converter":"org.apache.kafka.connect.storage.StringConverter", "key.template":"ID,NAME", "value.converter": "io.confluent.connect.avro.AvroConverter", "value.converter.schema.registry.url": "http://localhost:8081", "value.converter.schemas.enable": "false", "confluent.topic.bootstrap.servers":"localhost:9092", "oracle.server": "192.168.56.117", "oracle.port": "1522", "oracle.sid": "helowin", "oracle.username": "system", "oracle.password": "system", "start.from":"snapshot", "redo.log.topic.name": "SimpleOracleCDC_24", "redo.log.consumer.bootstrap.servers":"localhost:9092", "table.inclusion.regex":"helowin.SCOTT.TEST2", "numeric.mapping":"best_fit", "_table.topic.name.template_":"Using template vars to set change event topic for each table", "table.topic.name.template": "${databaseName}24.${schemaName}24.${tableName}24", "connection.pool.max.size": 20, "confluent.topic.replication.factor":1, "topic.creation.groups": "redo", "topic.creation.redo.include": "SimpleOracleCDC_24", "topic.creation.redo.replication.factor": 1, "topic.creation.redo.partitions": 1, "topic.creation.redo.cleanup.policy": "delete", "topic.creation.redo.retention.ms": 1209600000, "topic.creation.default.replication.factor": 1, "topic.creation.default.partitions": 1, "topic.creation.default.cleanup.policy": "compact" } }'

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V5ZW9mZWFnbGU_size_16_color_FFFFFF_t_70]

## 2.1, redo log对应的topic: 包含执行的sql语句 ##

"OPERATION": { 
        "string": "INSERT"
      },
      "OPERATION_CODE": { 
        "int": 1
      },
     "SQL_REDO": { 
        "string": "insert into \"SCOTT\".\"TEST2\"(\"ID\",\"NAME\",\"MARK\") values ('3','aa','ee');"
      },

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V5ZW9mZWFnbGU_size_16_color_FFFFFF_t_70 1]

## 2.2, record event 对应的topic: 一条表数据 ##

"ID": { 
        "bytes": "\u0003"
      },
      "NAME": { 
        "string": "aa"
      },
      "MARK": { 
        "string": "ee"
      },
      "op_type": { 
        "string": "D" // "U": update, "D":delete, "R":read
      },

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V5ZW9mZWFnbGU_size_16_color_FFFFFF_t_70 2]

# 3, oraclecdc-- > hbase #

*  oracle插入数据 ( 3,‘aa’, ‘ee’)  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V5ZW9mZWFnbGU_size_16_color_FFFFFF_t_70 3]

ksql> CREATE sink CONNECTOR `www-hbase-sink2` WITH(
        "connector.class"='io.svectors.hbase.sink.HBaseSinkConnector',
        "zookeeper.quorum"='192.168.56.161:2181',
        "event.parser.class"='io.svectors.hbase.parser.AvroEventParser',
        "hbase.table.name"='www',
        "hbase.www.rowkey.columns"='ID',
        "hbase.www.family"='f',
        "topics"='helowin24.SCOTT24.TEST224'
        );
    
    #查看hbase数据
    hbase(main):037:0> scan 'www'
    ROW                                      COLUMN+CELL
     java.nio.HeapByteBuffer[pos=0 lim=1 cap column=f:ID, timestamp=1626144115526, value=java.nio.HeapByteBuffer[pos=0 lim=1 cap=1]
     =1]
     java.nio.HeapByteBuffer[pos=0 lim=1 cap column=f:MARK, timestamp=1626144115526, value=ee
     =1]
     java.nio.HeapByteBuffer[pos=0 lim=1 cap column=f:NAME, timestamp=1626144115526, value=aa
     =1]
     java.nio.HeapByteBuffer[pos=0 lim=1 cap column=f:current_ts, timestamp=1626144115526, value=1626144115950
     =1]
    ...
    1 row(s) in 0.0060 seconds

[https_docs.confluent.io_kafka-connect-oracle-cdc_current_overview.html]: https://docs.confluent.io/kafka-connect-oracle-cdc/current/overview.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V5ZW9mZWFnbGU_size_16_color_FFFFFF_t_70]: /images/20221004/1aac0cde18d24431bb5d8ef042ec3d32.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V5ZW9mZWFnbGU_size_16_color_FFFFFF_t_70 1]: /images/20221004/d3615478493b45c6a545f3f1ae7b7fbc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V5ZW9mZWFnbGU_size_16_color_FFFFFF_t_70 2]: /images/20221004/da8764c54d47433b932fef5ab89247c3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V5ZW9mZWFnbGU_size_16_color_FFFFFF_t_70 3]: /images/20221004/977c98f030c74dc98d0feabf4078f28f.png