You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by GitBox <gi...@apache.org> on 2021/12/01 14:49:27 UTC
[GitHub] [incubator-doris] xiaokangguo commented on a change in pull request #7210: [feat](flink-connector) add BATCH DELETE support
xiaokangguo commented on a change in pull request #7210:
URL: https://github.com/apache/incubator-doris/pull/7210#discussion_r759973594
##########
File path: docs/en/extending-doris/flink-doris-connector.md
##########
@@ -132,71 +132,86 @@ INSERT INTO flink_doris_sink select name,age,price,sale from flink_doris_source
### DataStreamSink
+> Batch Delete (`MergeType.MERGE`) is only for the table of the unique key model, more details please refer to [Batch Delete](http://doris.incubator.apache.org/master/en/administrator-guide/load-data/batch-delete-manual.html)
+
+
+* Json data
+
```java
-// -------- sink with raw json string stream --------
Properties pro = new Properties();
-pro.setProperty("format", "json");
-pro.setProperty("strip_outer_array", "true");
-env.fromElements( "{\"longitude\": \"116.405419\", \"city\": \"北京\", \"latitude\": \"39.916927\"}")
- .addSink(
- DorisSink.sink(
- DorisReadOptions.builder().build(),
- DorisExecutionOptions.builder()
- .setBatchSize(3)
- .setBatchIntervalMs(0l)
- .setMaxRetries(3)
- .setStreamLoadProp(pro).build(),
- DorisOptions.builder()
- .setFenodes("FE_IP:8030")
- .setTableIdentifier("db.table")
- .setUsername("root")
- .setPassword("").build()
- ));
-
-OR
-env.fromElements("{\"longitude\": \"116.405419\", \"city\": \"北京\", \"latitude\": \"39.916927\"}")
- .addSink(
- DorisSink.sink(
- DorisOptions.builder()
- .setFenodes("FE_IP:8030")
- .setTableIdentifier("db.table")
- .setUsername("root")
- .setPassword("").build()
- ));
-
-
-// -------- sink with RowData stream --------
-DataStream<RowData> source = env.fromElements("")
- .map(new MapFunction<String, RowData>() {
- @Override
- public RowData map(String value) throws Exception {
- GenericRowData genericRowData = new GenericRowData(3);
- genericRowData.setField(0, StringData.fromString("北京"));
- genericRowData.setField(1, 116.405419);
- genericRowData.setField(2, 39.916927);
- return genericRowData;
- }
- });
+env.fromElements(
+ "{\"longitude\": \"116.405419\", \"city\": \"北京\", \"latitude\": \"39.916927\"}",
+ "{\"longitude\": \"116.405419\", \"city\": \"beijing\", \"latitude\": \"39.916927\"}"
+ )
+ .addSink(
+ DorisSink.sink(
+ DorisReadOptions.builder()
+ .build(),
+ DorisExecutionOptions.builder()
+ .setBatchSize(3)
+ .setBatchIntervalMs(0l)
+ .setMaxRetries(3)
+ .setStreamLoadProp(pro)
+ .setDataFormat(DorisExecutionOptions.DataFormat.JSON)
+ .setStripOuterArray(true)
+ .setMergeType(DorisExecutionOptions.MergeType.MERGE)
+ .setDeleteLabel("city='beijing'")
+ .build(),
+ DorisOptions.builder()
Review comment:
As a sink example,it might be better to tell the user how to make the data sink to doris directly by basic parameters.Since there are so many parameters.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org