ORACLE的DDL日志推送到Kafka，并接入Flink，进行统计

时间：2020-04-08 16:07:45 阅读：508 评论：0 收藏：0 [点我收藏+]

标签：stream row connector 模式 env cer ora return red

ORACLE的DDL日志推送到Kafka，并接入Flink，进行统计

本次测试的环境：

环境：docker oracle12c

日志模式：归档日志模式 archivelog

用户：scott/tiger 具有dba权限

大数据组件：kafka(默认创建好topic：flink_topic)，zookeeper

额外组件：kafka-connect-oracle-1.0.jar

下载地址： https://github.com/erdemcer/kafka-connect-oracle

1. 创建测试表，并插入几条记录

技术图片

2.开启归档日志模式

sqlplus / as sysdba
SQL> shutdown immediate
SQL> startup mount
SQL> alter database archivelog;
SQL> alter database open;
SQL> alter database add supplemental log data (all) columns;
SQL> conn username/password

3.准备相关Jar包

1. 从https://github.com/erdemcer/kafka-connect-oracle下载整个项目，把整个项目mvn clean package成kafa-connect-oracle-1.0.jar
2. 下载一个oracle的jdbc驱动jar—ojdbc7.jar
3. 将kafa-connect-oracle-1.0.jar and ojdbc7.jar放在kafka的安装包下的lib目录下
4. 将github项目里面的config/OracleSourceConnector.properties文件拷贝到kafak/config

4. 配置相关文件

# vi /opt/cloudera/parcels/KAFKA/lib/kafka/config/OracleSourceConnector.properties

技术图片

# vi /opt/cloudera/parcels/KAFKA/lib/kafka/config/connect-standalone.properties

技术图片

5. 运行启动Connector

bin/connect-standalone.sh config/connect-standalone.properties config/OracleSourceConnector.properties

6. 启动consumer

bin/kafka-console-consumer.sh --bootstrap-server 192.168.58.177:9092 --from-beginning --topic flink_topic

7. 最后结果

{"schema":{"type":"struct","fields":
[
    {"type":"int64","optional":false,"field":"SCN"},
    {"type":"string","optional":false,"field":"SEG_OWNER"},
    {"type":"string","optional":false,"field":"TABLE_NAME"},
    {"type":"int64","optional":false,"name":"org.apache.kafka.connect.data.Timestamp","version":1,"field":"TIMESTAMP"},
    {"type":"string","optional":false,"field":"SQL_REDO"},
    {"type":"string","optional":false,"field":"OPERATION"},
    {"type":"struct","fields":
        [
            {"type":"double","optional":true,"field":"ID"},
            {"type":"string","optional":true,"field":"NAME"}
        ],"optional":true,"name":"value","field":"data"},
    {"type":"struct","fields":
        [
            {"type":"double","optional":true,"field":"ID"},
            {"type":"string","optional":true,"field":"NAME"}
        ],"optional":true,"name":"value","field":"before"}
],"optional":false,"name":"flink_test.scott.test_20200402.row"},
"payload":
{
    "SCN":1719627,
    "SEG_OWNER":"SCOTT",
    "TABLE_NAME":"TEST_20200402",
    "TIMESTAMP":1585773001000,
    "SQL_REDO":"insert into \"SCOTT\".\"TEST_20200402\"(\"ID\",\"NAME\") values (2,‘zhangfei‘)",
    "OPERATION":"INSERT",
    "data":{"ID":2.0,"NAME":"zhangfei"},
    "before":null
}
}

8.Flink中读取数据 Demo

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment Env = StreamExecutionEnvironment.getExecutionEnvironment();

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "192.168.58.177:9092");
        properties.setProperty("zookeeper.connect", "192.168.58.171:2181,192.168.58.177:2181");
        properties.setProperty("group.id", "test");

        FlinkKafkaConsumer myConsumer = new FlinkKafkaConsumer("flink_topic",new SimpleStringSchema(),properties);


        //设置并行度
        myConsumer.setStartFromEarliest();

        //添加数据源,json格式
        DataStreamSource<ObjectNode> stream = Env.addSource(myConsumer);

        stream.print();

        Env.execute("flink_topic");
    }


    public static class DataS{

        public Integer id;
        public String name;

        public Integer getId() {
            return id;
        }

        public void setId(Integer id) {
            this.id = id;
        }

        public String getName() {
            return name;
        }

        public void setName(String name) {
            this.name = name;
        }

    }