实时数仓：基于 Flink CDC 实现 Oracle 数据实时更新到 Kudu

# 启动systemctl start docker
# 设置开机启动systemctl enable docker.service
# 查看 docker 服务状态systemctl status docker

# 查找 Oracle 镜像版本docker search oracle
# 下载相对应镜像，这里我们下载 truevoly/oracle-12c 版本docker pull truevoly/oracle-12c
# 运行 Docker 容器docker run -d -p 1521:1521 --name oracle12c truevoly/oracle-12c
# 进入容器docker exec -it oracle12c /bin/bash

-- 如有必要重新 source 一下 .profile 文件source /home/oracle/.profile
-- 1. 切换到 Oracle 用户su oracle
-- 2. 以 DBA 身份连接数据库$ORACLE_HOME/bin/sqlplus /nologconn /as sysdbashow user
-- 3. 启用日志归档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;
-- 4. 检查日志是否归档archive log list;

CREATE TABLESPACE logminer_tbs DATAFILE '/opt/oracle/oradata/SID/logminer_tbs.dbf' SIZE 25M REUSE AUTOEXTEND ON MAXSIZE UNLIMITED;

CREATE USER flinkuser IDENTIFIED BY flinkpw DEFAULT TABLESPACE LOGMINER_TBS QUOTA UNLIMITED ON LOGMINER_TBS;GRANT CREATE SESSION TO flinkuser;GRANT SET CONTAINER TO flinkuser;GRANT SELECT ON V_$DATABASE to flinkuser;GRANT FLASHBACK ANY TABLE TO flinkuser;GRANT SELECT ANY TABLE TO flinkuser;GRANT SELECT_CATALOG_ROLE TO flinkuser;GRANT EXECUTE_CATALOG_ROLE TO flinkuser;GRANT SELECT ANY TRANSACTION TO flinkuser;GRANT LOGMINING TO flinkuser;GRANT CREATE TABLE TO flinkuser;GRANT LOCK ANY TABLE TO flinkuser;GRANT ALTER ANY TABLE TO flinkuser;GRANT CREATE SEQUENCE TO flinkuser;GRANT EXECUTE ON DBMS_LOGMNR TO flinkuser;GRANT EXECUTE ON DBMS_LOGMNR_D TO flinkuser;GRANT SELECT ON V_$LOG TO flinkuser;GRANT SELECT ON V_$LOG_HISTORY TO flinkuser;GRANT SELECT ON V_$LOGMNR_LOGS TO flinkuser;GRANT SELECT ON V_$LOGMNR_CONTENTS TO flinkuser;GRANT SELECT ON V_$LOGMNR_PARAMETERS TO flinkuser;GRANT SELECT ON V_$LOGFILE TO flinkuser;GRANT SELECT ON V_$ARCHIVED_LOG TO flinkuser;GRANT SELECT ON V_$ARCHIVE_DEST_STATUS TO flinkuser;

-- 创建 Oracle 表，用做 Source 端CREATE TABLE FLINKUSER.TEST1(  ID   NUMBER(10,0) NOT NULL ENABLE,    NAME VARCHAR2(50),   PRIMARY KEY(ID)  ) TABLESPACE LOGMINER_TBS;-- 手动插入几条数据INSERT INTO FLINKUSER.TEST1 (ID,NAME) VALUES (1,'1111');

-- 对数据库配置ALTER DATABASE ADD SUPPLEMENTAL LOG DATA;-- 对表进行配置ALTER TABLE FLINKUSER.TEST1 ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;

<dependency>   <groupId>com.ververica</groupId>   <artifactId>flink-connector-oracle-cdc</artifactId>   <version>2.2-SNAPSHOT</version>   <!-- 此处依赖需要设置为 scope，其他 flink 依赖需设置为 provied，Oceanus 平台已提供 -->   <scope>compile</scope></dependency>

package com.demo;
import com.ververica.cdc.connectors.oracle.OracleSource;import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.streaming.api.functions.source.SourceFunction;import org.apache.flink.table.api.EnvironmentSettings;import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
public class OracleToKudu {    public static void main(String[] args) throws Exception {        EnvironmentSettings settings = EnvironmentSettings                .newInstance()                .useBlinkPlanner()                .inStreamingMode()                .build();
        StreamExecutionEnvironment sEnv = StreamExecutionEnvironment.getExecutionEnvironment();        StreamTableEnvironment tEnv = StreamTableEnvironment.create(sEnv,settings);
        // SQL 写法        tEnv.executeSql("CREATE TABLE `oracleSource` (\n" +                " ID      BIGINT,\n" +                " NAME    VARCHAR,\n" +                " PRIMARY KEY(ID) NOT ENFORCED )\n" +                " WITH (\n" +                "  'connector' = 'oracle-cdc',\n" +                // 请修改成 Oracle 所在的实际 IP 地址                "  'hostname' = 'xx.xx.xx.xx',\n" +                "  'port' = '1521',\n" +                "  'username' = 'flinkuser',\n" +                "  'password' = 'flinkpw',\n" +                "  'database-name' = 'xe',\n" +                "  'schema-name' = 'flinkuser',\n" +                "  'table-name' = 'test1'\n" +                ")");
        // Stream API 写法        // SourceFunction<String> sourceFunction = OracleSource.<String>builder()        //         .hostname("xx.xx.xx.xx")        //         .port(1521)        //         .database("xe")        //         .schemaList("flinkuser")        //         .tableList("flinkuser.test1")        //         .username("flinkuser")        //         .password("flinkpw")        //         .deserializer(new JsonDebeziumDeserializationSchema())        //         .build();        // sEnv.addSource(sourceFunction)

        tEnv.executeSql("CREATE TABLE `kudu_sink_table` (\n" +                " `id`    BIGINT,\n" +                " `name`  VARCHAR\n" +                ") WITH (\n" +                " 'connector.type' = 'kudu',\n" +                // 请修改成实际的 master IP 地址                " 'kudu.masters' = 'master-01:7051,master-02:7051,master-03:7051',\n" +                " 'kudu.table' = 'JoylyuTest1',\n" +                " 'kudu.hash-columns' = 'id',\n" +                " 'kudu.primary-key-columns' = 'id'\n" +                ")");
        // 笔者这里只是进行了最简化的数据转移功能，请根据实际业务情况进行开发        tEnv.executeSql("insert into kudu_sink_table select * from oracleSource");
    }}

# 进入 kudu 目录下cd /usr/local/service/kudu/bin
# 查看集群所有表./kudu table list master-01,master-02,master-03
# 查询 JoylyuTest1 表的数据./kudu table scan master-01,master-02,master-03 JoylyuTest1

CREATE EXTERNAL TABLE ImpalaJoylyuTest1STORED AS KUDUTBLPROPERTIES (  'kudu.master_addresses' = 'master-01:7051,master-02:7051,master-03:7051',   'kudu.table_name' = 'JoylyuTest1');

Caused by: io.debezium.DebeziumException: Supplemental logging not configured for table HELOWIN.FLINKUSER.test1.  Use command: ALTER TABLE FLINKUSER.test1 ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS  at io.debezium.connector.oracle.logminer.LogMinerHelper.checkSupplementalLogging(LogMinerHelper.java:407)  at io.debezium.connector.oracle.logminer.LogMinerStreamingChangeEventSource.execute(LogMinerStreamingChangeEventSource.java:132)  ... 7 more

static String tableSupplementalLoggingCheckQuery(TableId tableId) {        return String.format("SELECT 'KEY', LOG_GROUP_TYPE FROM %s WHERE OWNER = '%s' AND TABLE_NAME = '%s'", ALL_LOG_GROUPS, tableId.schema(), tableId.table());    }

// 将表名转为小写的源代码如下private TableId toLowerCaseIfNeeded(TableId tableId) {    return tableIdCaseInsensitive ? tableId.toLowercase() : tableId;}

'debezium.log.mining.strategy'='online_catalog','debezium.log.mining.continuous.mine'='true'

Exception in thread "main" java.lang.IllegalArgumentException: The parallelism of non parallel operator must be 1.  at org.apache.flink.util.Preconditions.checkArgument(Preconditions.java:138)  at org.apache.flink.api.common.operators.util.OperatorValidationUtils.validateParallelism(OperatorValidationUtils.java:35)  at org.apache.flink.streaming.api.datastream.DataStreamSource.setParallelism(DataStreamSource.java:114)  at com.demo.OracleToKudu.main(OracleToKudu.java:67)

// 报错代码部分public static void validateParallelism(int parallelism, boolean canBeParallel) {    Preconditions.checkArgument(canBeParallel || parallelism == 1, "The parallelism of non parallel operator must be 1.");    Preconditions.checkArgument(parallelism > 0 || parallelism == -1, "The parallelism of an operator must be at least 1, or ExecutionConfig.PARALLELISM_DEFAULT (use system default).");}