This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-doris.git
The following commit(s) were added to refs/heads/master by this push:
new cacc294 [demo] Add Flink oracle cdc demo (#7845)
cacc294 is described below
commit cacc29470be39636bfe4f8cf2c645e56c02d342a
Author: caoliang-web <[email protected]>
AuthorDate: Mon Jan 31 22:11:38 2022 +0800
[demo] Add Flink oracle cdc demo (#7845)
---
samples/doris-demo/flink-demo/pom.xml | 6 +
.../doris/demo/flink/cdc/FlinkOracleCdcDemo.java | 130 +++++++++++++++++++++
.../cdc/JsonDebeziumDeserializationSchema.java | 76 ++++++++++++
3 files changed, 212 insertions(+)
diff --git a/samples/doris-demo/flink-demo/pom.xml
b/samples/doris-demo/flink-demo/pom.xml
index ea664d9..2e751ea 100644
--- a/samples/doris-demo/flink-demo/pom.xml
+++ b/samples/doris-demo/flink-demo/pom.xml
@@ -111,6 +111,12 @@ under the License.
<artifactId>slf4j-simple</artifactId>
<version>1.7.25</version>
</dependency>
+ <!--
https://mvnrepository.com/artifact/com.ververica/flink-connector-oracle-cdc -->
+ <dependency>
+ <groupId>com.ververica</groupId>
+ <artifactId>flink-connector-oracle-cdc</artifactId>
+ <version>2.1.1</version>
+ </dependency>
</dependencies>
<build>
<plugins>
diff --git
a/samples/doris-demo/flink-demo/src/main/java/org/apache/doris/demo/flink/cdc/FlinkOracleCdcDemo.java
b/samples/doris-demo/flink-demo/src/main/java/org/apache/doris/demo/flink/cdc/FlinkOracleCdcDemo.java
new file mode 100644
index 0000000..5a1045a
--- /dev/null
+++
b/samples/doris-demo/flink-demo/src/main/java/org/apache/doris/demo/flink/cdc/FlinkOracleCdcDemo.java
@@ -0,0 +1,130 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+package org.apache.doris.demo.flink.cdc;
+
+import com.alibaba.fastjson.JSONObject;
+import com.ververica.cdc.connectors.oracle.OracleSource;
+import org.apache.doris.flink.cfg.DorisExecutionOptions;
+import org.apache.doris.flink.cfg.DorisOptions;
+import org.apache.doris.flink.cfg.DorisReadOptions;
+import org.apache.doris.flink.cfg.DorisSink;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.environment.CheckpointConfig;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.types.logical.DoubleType;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.VarCharType;
+
+import java.util.Properties;
+
+/**
+ *Realize the consumption of oracle log data through flink cdc,
+ * and then import oracle data into doris table data in real time
+ * through the flink doris connector RowData data stream;
+ */
+public class FlinkOracleCdcDemo {
+
+ private static final String DATABASE_NAME = "xxx";
+
+ private static final String HOST_NAME = "127.0.0.1";
+
+ private static final int PORT = 1521;
+
+ private static final String SCHEMA_NAME = "xxx";
+
+ private static final String TABLE_NAME = "schema_name.table_name";
+
+ private static final String USER_NAME = "xxx";
+
+ private static final String PASSWORD = "xxx";
+
+
+ public static void main(String[] args) throws Exception {
+
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ env.enableCheckpointing(10000);
+
env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
+
+ Properties props = new Properties();
+
props.setProperty("debezium.database.tablename.case.insensitive","false");
+ props.setProperty("debezium.log.mining.strategy","online_catalog");
+ props.setProperty("debezium.log.mining.continuous.mine","true");
+
+
+ SourceFunction<JSONObject> build = OracleSource.<JSONObject>builder()
+ .database(DATABASE_NAME)
+ .hostname(HOST_NAME)
+ .port(PORT)
+ .username(USER_NAME)
+ .password(PASSWORD)
+ .schemaList(SCHEMA_NAME)
+ .tableList(TABLE_NAME)
+ .debeziumProperties(props)
+ .deserializer(new JsonDebeziumDeserializationSchema())
+ .build();
+
+ DataStreamSource<JSONObject> dataStreamSource = env.addSource(build);
+
+
+ SingleOutputStreamOperator<RowData> map = dataStreamSource.map(new
MapFunction<JSONObject, RowData>() {
+ @Override
+ public RowData map(JSONObject jsonObject) throws Exception {
+ GenericRowData genericRowData = new GenericRowData(4);
+ genericRowData.setField(0,
Integer.valueOf(jsonObject.getInteger("id")));
+ genericRowData.setField(1,
StringData.fromString(jsonObject.getString("name")));
+ genericRowData.setField(2,
StringData.fromString(jsonObject.getString("description")));
+ genericRowData.setField(3,
Double.valueOf(jsonObject.getDoubleValue("weight")));
+ return genericRowData;
+ }
+ });
+
+
+ String[] fields = {"id", "name", "description","weight"};
+
+ LogicalType[] types={new IntType(),new VarCharType(),new
VarCharType(), new DoubleType()};
+
+ Properties pro = new Properties();
+ pro.setProperty("format", "json");
+ pro.setProperty("strip_outer_array", "false");
+
+ map.addSink(
+ DorisSink.sink(
+ fields,
+ types,
+ DorisReadOptions.builder().build(),
+ DorisExecutionOptions.builder()
+ .setBatchSize(3)
+ .setMaxRetries(3)
+
+ .build(),
+ DorisOptions.builder()
+ .setFenodes("127.0.0.1:8030")
+ .setTableIdentifier("db_name.table_name")
+ .setUsername("root")
+ .setPassword("").build()
+ ));
+
+ env.execute("flink oracle cdc");
+ }
+}
diff --git
a/samples/doris-demo/flink-demo/src/main/java/org/apache/doris/demo/flink/cdc/JsonDebeziumDeserializationSchema.java
b/samples/doris-demo/flink-demo/src/main/java/org/apache/doris/demo/flink/cdc/JsonDebeziumDeserializationSchema.java
new file mode 100644
index 0000000..5b714b6
--- /dev/null
+++
b/samples/doris-demo/flink-demo/src/main/java/org/apache/doris/demo/flink/cdc/JsonDebeziumDeserializationSchema.java
@@ -0,0 +1,76 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+package org.apache.doris.demo.flink.cdc;
+
+import com.alibaba.fastjson.JSONObject;
+import com.ververica.cdc.debezium.DebeziumDeserializationSchema;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.util.Collector;
+import org.apache.kafka.connect.data.Field;
+import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.source.SourceRecord;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.Locale;
+
+/**
+ *Customize DebeziumDeserializationSchema, return jsonObject
+ */
+public class JsonDebeziumDeserializationSchema implements
DebeziumDeserializationSchema<JSONObject> {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(JsonDebeziumDeserializationSchema.class);
+
+ private static final long serialVersionUID = 7906905121308228264L;
+
+ public JsonDebeziumDeserializationSchema() {
+ }
+ @Override
+ public void deserialize(SourceRecord sourceRecord, Collector<JSONObject>
collector) throws Exception {
+ JSONObject resJson = new JSONObject();
+ try {
+ Struct valueStruct = (Struct) sourceRecord.value();
+ Struct afterStruct = valueStruct.getStruct("after");
+ Struct beforeStruct = valueStruct.getStruct("before");
+ if (afterStruct != null && beforeStruct != null) {
+ LOGGER.info("Updated, ignored ...");
+ } else if (afterStruct != null) {
+ List<Field> fields = afterStruct.schema().fields();
+ String name;
+ Object value;
+ for (Field field : fields) {
+ name = field.name();
+ value = afterStruct.get(name);
+ resJson.put(name.toLowerCase(Locale.ROOT), value);
+ }
+ } else if (beforeStruct != null) {
+ LOGGER.info("Deleted, ignored ...");
+ } else {
+ LOGGER.warn("No this operation ...");
+ }
+ } catch (Exception e) {
+ LOGGER.error("Deserialize throws exception:", e);
+ }
+ collector.collect(resJson);
+ }
+
+ @Override
+ public TypeInformation<JSONObject> getProducedType() {
+ return BasicTypeInfo.of(JSONObject.class);
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
