morningman commented on a change in pull request #7210: URL: https://github.com/apache/incubator-doris/pull/7210#discussion_r763666319
########## 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: Oh I see, could you please contact me with Wechat(morningman-cmy). There are some details need to be discussed. -- 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-unsubscr...@doris.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org For additional commands, e-mail: commits-h...@doris.apache.org
