flink doris批量sink
idea maven 依赖
<dependency>
<groupId>org.apache.doris</groupId>
<artifactId>flink-doris-connector-1.11_2.12</artifactId>
<version>1.0.3</version>
</dependency>
val properties = Properties()
properties.setProperty("format", "json")
properties.setProperty("strip_outer_array", "true")
properties.setProperty("read_json_by_line", "true")
val executionOptions = DorisExecutionOptions.builder()
.setBatchSize(20000)
.setBatchIntervalMs(10000L)
.setEnableDelete(false)
.setMaxRetries(3)
.setStreamLoadProp(properties)
.build()
data.addSink(
DorisSink.sink(
executionOptions,
DorisOptions(
"host:8030",
"root",
"password",
"db.table"
)
)
)
.uid("sink-doris")
.name("数据写入doris")