Flink CDC + Oracle Demo

本文用于说明Flink集成oracle cdc的方式

pom.xml

			org.apache.flink
			flink-java
			${flink.version}

			
		
		
			org.apache.flink
			flink-clients_2.11
			${flink.version}
		
		
			com.ververica
			flink-connector-oracle-cdc

			
			2.1.1
		
		
			org.apache.flink
			flink-connector-jdbc_2.11
			${flink.version}
		
		
			org.apache.flink
			flink-table-planner-blink_${scala.version}
			${flink.version}
		
		
			org.apache.flink
			flink-streaming-java_${scala.version}
			${flink.version}
		

		
			org.apache.flink
			flink-clients_${scala.version}
			${flink.version}
		
		
			org.apache.flink
			flink-cep_${scala.version}
			${flink.version}
		
		
			org.apache.flink
			flink-json
			${flink.version}
		
		
			com.alibaba
			fastjson
			1.2.51
		

public static void main(String[] args) throws Exception {
        Properties props = new Properties();
        props.setProperty("database.tablename.case.insensitive", "false");
        props.setProperty("database.connection.adapter", "logminer");
        // 要同步快，这个配置必须加，不然非常慢
        props.setProperty("log.mining.strategy", "online_catalog");
        props.setProperty("log.mining.continuous.mine", "true");
        SourceFunction sourceFunction = OracleSource.builder().hostname("ip").port(1521).database("ORCL")
            .schemaList("schema")
            .tableList("schema.table")
            .debeziumProperties(props).username("username").password("password").deserializer(new JsonDebeziumDeserializationSchema()) 
            .build();

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.addSource(sourceFunction).process(new Process()).setParallelism(1);

        env.execute("OracleSource Demo");
    }

public class Process extends ProcessFunction {
    private static AtomicInteger count = new AtomicInteger();
    @Override
    public void processElement(String s, ProcessFunction.Context context, Collector