[cdc-pipeline-connector][doris] Introduce Doris cdc pipeline DataSink #2729

JNSimba · 2023-11-22T03:28:01Z

This closes #2646
add a DorisDataSink that implement interface of DataSink to build a pipeline.

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connectors-doris/pom.xml

...ne-connectors-doris/src/main/java/com/ververica/cdc/connectors/doris/sink/DorisDataSink.java

lvyanquan · 2023-11-22T08:58:33Z

...ectors-doris/src/main/java/com/ververica/cdc/connectors/doris/sink/DorisMetadataApplier.java

+    private void applyAddColumnEvent(AddColumnEvent event) throws IOException, IllegalArgumentException {
+        TableId tableId = event.tableId();
+        List<AddColumnEvent.ColumnWithPosition> addedColumns = event.getAddedColumns();
+        for(AddColumnEvent.ColumnWithPosition col: addedColumns){


There are some kinds of ColumnPosition, Does here deal with LAST type only?

Yes, because adding a value column in Doris does not support adding it to the middle of multiple Key columns.
At the same time, the default is to use JSON format to import. The order of the columns will not affect the data quality. The value column can be directly appended to the end.

...ectors-doris/src/main/resources/META-INF/services/com.ververica.cdc.common.factories.Factory

...ectors-doris/src/main/java/com/ververica/cdc/connectors/doris/sink/DorisDataSinkOptions.java

...ectors-doris/src/main/java/com/ververica/cdc/connectors/doris/sink/DorisMetadataApplier.java

...ne-connectors-doris/src/main/java/com/ververica/cdc/connectors/doris/sink/DorisDatabase.java

# Conflicts: # flink-cdc-common/src/main/java/com/ververica/cdc/common/schema/Schema.java

JNSimba · 2023-11-23T09:45:07Z

Addressed the comments. PTAL @lvyanquan

gtk96 · 2023-11-24T03:22:06Z

@JNSimba Run 'mvn spotless:apply' to fix these violations.

JNSimba · 2023-11-24T06:02:43Z

@JNSimba Run 'mvn spotless:apply' to fix these violations.

Thanks for the reminder, it has been processed @gtk96

flink-cdc-common/src/main/java/com/ververica/cdc/common/data/GenericStringData.java

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-doris/pom.xml

...tor-doris/src/main/java/com/ververica/cdc/connectors/doris/factory/DorisDataSinkFactory.java

lvyanquan · 2023-11-27T10:40:11Z

#2734 removes GenericRecordData and adds BinaryRecordData, so we can get Objects directly from RecordData and need to use FieldGetter. I've raised a pr to give an example to use it in ValuesDataSink.

lvyanquan · 2023-11-27T12:28:13Z

...nector-doris/src/main/java/com/ververica/cdc/connectors/doris/sink/DorisEventSerializer.java

+/** A serializer for Event to Tuple2<String, byte[]> */
+public class DorisEventSerializer implements DorisRecordSerializer<Event> {
+    private ObjectMapper objectMapper = new ObjectMapper();
+    private Map<TableId, Schema> schemaMaps = new HashMap<>();


A troublesome problem is that we need to maintain the schemaMaps in State to recover from failure, so we need to add a subclass of DorisWriter or DorisBatchWriter to overwrite snapshotState method.
What do you think?

Sorry for this comment, as the discuss before, CreateTableEvent will always sent before DataChangEvent, so we don't need to consider this situation.

Yes, I understand that it will resend CreateTableEvent even if it is restarted.

# Conflicts: # flink-cdc-common/src/main/java/com/ververica/cdc/common/data/GenericStringData.java

flink-cdc-common/src/main/java/com/ververica/cdc/common/utils/RecordDataUtils.java

# Conflicts: # flink-cdc-common/src/main/java/com/ververica/cdc/common/utils/SchemaUtils.java

...nector-doris/src/main/java/com/ververica/cdc/connectors/doris/sink/DorisEventSerializer.java

# Conflicts: # flink-cdc-common/src/main/java/com/ververica/cdc/common/utils/SchemaUtils.java

JNSimba · 2023-11-29T12:19:59Z

Addressed the comments. PTAL @lvyanquan

lvyanquan · 2023-11-30T07:36:46Z

Thanks for your contribution. Over look good to me. Left some comments, and can you clean the commit message？

flink-cdc-common/src/main/java/com/ververica/cdc/common/types/utils/DataTypeUtils.java

...nector-doris/src/main/resources/META-INF/services/com.ververica.cdc.common.factories.Factory

...tor-doris/src/main/java/com/ververica/cdc/connectors/doris/factory/DorisDataSinkFactory.java

...nector-doris/src/main/java/com/ververica/cdc/connectors/doris/sink/DorisDataSinkOptions.java

...nector-doris/src/main/java/com/ververica/cdc/connectors/doris/sink/DorisMetadataApplier.java

JNSimba · 2023-11-30T09:59:08Z

Addressed the comments. PTAL @lvyanquan

...nector-doris/src/main/java/com/ververica/cdc/connectors/doris/sink/DorisDataSinkOptions.java

JNSimba · 2023-12-01T09:39:33Z

Addressed the comments. PTAL @lvyanquan

leonardBang · 2023-12-05T07:22:10Z

Resolved by 4abd86a

wudi added 2 commits November 22, 2023 11:23

add doris sink connector for pipeline

38912d1

fix

49befdf

lvyanquan reviewed Nov 22, 2023

View reviewed changes

wudi added 5 commits November 23, 2023 14:33

Merge branch 'master' into doris-sink

68a92e7

# Conflicts: # flink-cdc-common/src/main/java/com/ververica/cdc/common/schema/Schema.java

remove serialiazer and typemapping

44f9ce2

code style format

5addc14

add doc

808efb4

add other options

26103b7

code format

f5bccdb

lvyanquan reviewed Nov 27, 2023

View reviewed changes

Merge branch 'master' into doris-sink

5cf8dc3

# Conflicts: # flink-cdc-common/src/main/java/com/ververica/cdc/common/data/GenericStringData.java

github-actions bot added common build labels Nov 28, 2023

wudi added 4 commits November 28, 2023 15:08

change identifier

a3593b6

Merge branch 'master' into doris-sink

103102c

update

9df7fef

remove GenericStringdata

86e2fea

lvyanquan reviewed Nov 28, 2023

View reviewed changes

flink-cdc-common/src/main/java/com/ververica/cdc/common/utils/RecordDataUtils.java Outdated Show resolved Hide resolved

wudi added 2 commits November 28, 2023 21:52

Merge branch 'master' into doris-sink

23d24eb

# Conflicts: # flink-cdc-common/src/main/java/com/ververica/cdc/common/utils/SchemaUtils.java

update pom and remove recorddatautil

e91ee2c

lvyanquan reviewed Nov 29, 2023

View reviewed changes

...nector-doris/src/main/java/com/ververica/cdc/connectors/doris/sink/DorisEventSerializer.java Outdated Show resolved Hide resolved

wudi added 4 commits November 29, 2023 16:30

Merge branch 'master' into doris-sink

82c6785

# Conflicts: # flink-cdc-common/src/main/java/com/ververica/cdc/common/utils/SchemaUtils.java

update

6ee36cd

code style

04bac72

fix bugs

2535767

schemachange fail throw exception

4e3ab7c

lvyanquan reviewed Nov 30, 2023

View reviewed changes

...nector-doris/src/main/java/com/ververica/cdc/connectors/doris/sink/DorisDataSinkOptions.java Outdated Show resolved Hide resolved

wudi added 2 commits November 30, 2023 16:29

add require option and code style

d5ae65c

update default option

154f080

lvyanquan reviewed Nov 30, 2023

View reviewed changes

...nector-doris/src/main/java/com/ververica/cdc/connectors/doris/sink/DorisMetadataApplier.java Outdated Show resolved Hide resolved

improve

2240520

JNSimba mentioned this pull request Dec 1, 2023

[docs] introduces of Doris Pipeline Connector #2788

Merged

lvyanquan reviewed Dec 1, 2023

View reviewed changes

...nector-doris/src/main/java/com/ververica/cdc/connectors/doris/sink/DorisDataSinkOptions.java Outdated Show resolved Hide resolved

remove unuse options

d09375b

JNSimba changed the title ~~[WIP][3.0][cdc-runtime] add DorisDataSink for Pipeline~~ [cdc-pipeline-connector][doris] Introduce Doris cdc pipeline DataSink Dec 2, 2023

wudi and others added 2 commits December 3, 2023 21:13

update pom release

f5f982a

[pipeline-connector][doris] support local time zone type. (#1)

0f272f4

lvyanquan mentioned this pull request Dec 4, 2023

[pipeline-connector][doris] add doris pipeline connector. #2810

Merged

leonardBang closed this Dec 5, 2023

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[cdc-pipeline-connector][doris] Introduce Doris cdc pipeline DataSink #2729

[cdc-pipeline-connector][doris] Introduce Doris cdc pipeline DataSink #2729

JNSimba commented Nov 22, 2023

lvyanquan Nov 22, 2023

JNSimba Nov 23, 2023

JNSimba commented Nov 23, 2023

gtk96 commented Nov 24, 2023

JNSimba commented Nov 24, 2023

lvyanquan commented Nov 27, 2023

lvyanquan Nov 27, 2023

lvyanquan Nov 28, 2023

JNSimba Nov 28, 2023

JNSimba commented Nov 29, 2023

lvyanquan commented Nov 30, 2023 •

edited

Loading

JNSimba commented Nov 30, 2023

JNSimba commented Dec 1, 2023

leonardBang commented Dec 5, 2023

[cdc-pipeline-connector][doris] Introduce Doris cdc pipeline DataSink #2729

[cdc-pipeline-connector][doris] Introduce Doris cdc pipeline DataSink #2729

Conversation

JNSimba commented Nov 22, 2023

lvyanquan Nov 22, 2023

Choose a reason for hiding this comment

JNSimba Nov 23, 2023

Choose a reason for hiding this comment

JNSimba commented Nov 23, 2023

gtk96 commented Nov 24, 2023

JNSimba commented Nov 24, 2023

lvyanquan commented Nov 27, 2023

lvyanquan Nov 27, 2023

Choose a reason for hiding this comment

lvyanquan Nov 28, 2023

Choose a reason for hiding this comment

JNSimba Nov 28, 2023

Choose a reason for hiding this comment

JNSimba commented Nov 29, 2023

lvyanquan commented Nov 30, 2023 • edited Loading

JNSimba commented Nov 30, 2023

JNSimba commented Dec 1, 2023

leonardBang commented Dec 5, 2023

lvyanquan commented Nov 30, 2023 •

edited

Loading