ORACLE的DDL日志 推送到Kafka，並用Flink進行實時計算

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;
        }

    }

 

 9. 運用Flink Sql進行實時計算

package com.flink;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.StreamTableEnvironment;

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

        //創建flink運行環境
        StreamExecutionEnvironment Env = StreamExecutionEnvironment.getExecutionEnvironment();

        //創建tableEnvironment
        StreamTableEnvironment TableEnv =  StreamTableEnvironment.create(Env);

        TableEnv.sqlUpdate("CREATE TABLE user_log2 (\n" +
                "    payload ROW(SCN string,SEG_OWNER string,data ROW(ID DECIMAL,NAME string))\n" +
                ") WITH (\n" +
                "    'connector.type' = 'kafka',\n" +
                "    'connector.version' = 'universal',\n" +
                "    'connector.topic' = 'flink_topic',\n" +
                "    'connector.startup-mode' = 'latest-offset',\n" +
                "    'connector.properties.group.id' = 'test',\n" +
                "    'connector.properties.zookeeper.connect' = '192.168.58.171:2181,192.168.58.177:2181,192.168.58.178:2181',\n" +
                "    'connector.properties.bootstrap.servers' = '192.168.58.177:9092',\n" +
                "    'format.type' = 'json',\n" +
                "    'format.json-schema' =\n" +
                "    '{\n" +
                "        type : \"object\",\n" +
                "        \"properties\":\n" +
                "        {\n" +
                "            \"payload\": {type : \"object\",\n" +
                "                \"properties\": \n" +
                "                 {\n" +
                "                    \"SCN\" : {type :\"string\"},\n" +
                "                    \"SEG_OWNER\" : {type :\"string\"},\n" +
                "                    \"data\": {type : \"object\",\n" +
                "                    \"properties\": {\"ID\": {type : \"integer\"},\n" +
                "                                   \"NAME\": {type : \"string\"}\n" +
                "                                  }\n" +
                "                             }\n" +
                "                 }}\n" +
                "        }\n" +
                "    }'\n" +
                ")"
        ) ;

        Table result=TableEnv.sqlQuery("select payload.data.NAME,sum(payload.data.ID) from user_log2 group by payload.data.NAME");

        TableEnv.toRetractStream(result,Types.TUPLE(Types.STRING,Types.STRING,Types.BIG_DEC,Types.STRING)) .print();
        Env.execute("flink job");
    }
}

運行結果：