doris版本是1.1

之前使用：

依赖

<dependency>
	<groupId>org.apache.doris</groupId>
	<artifactId>flink-doris-connector-1.14_2.12</artifactId>
	<version>1.1.0</version>
</dependency>

代码片段

//doris sink option
DorisSink.Builder<RowData> dorisSinkBuilder = DorisSink.builder();
DorisOptions.Builder dorisBuilder = DorisOptions.builder().setFenodes("xx.xx.xx.xx:xxxx")
		.setTableIdentifier(String.format("%s.%s", database, tableName))
		.setUsername("xxxx")
		.setPassword("xxxxxx");

// json format to streamload
Properties properties = new Properties();
properties.setProperty("format", "json");
properties.setProperty("read_json_by_line", "true");
DorisExecutionOptions.Builder  executionBuilder = DorisExecutionOptions.builder();
executionBuilder.setLabelPrefix("label-doris-"+TimeUtil.getDateStrByFormat(System.currentTimeMillis(), "yyyyMMdd_HHmm")).setStreamLoadProp(properties);

//通过Doris建表语句解析得来
List<SchemaPo> schemaPoList = DorisDDLService.getSchemaPoList(database, tableName);
String[] fields = new String[schemaPoList.size()];
DataType[] types = new DataType[schemaPoList.size()];
for (int i = 0; i < schemaPoList.size(); i++) {
	SchemaPo schemaPo = schemaPoList.get(i);
	fields[i] = schemaPo.getSignal(); //字段数组
	types[i] = DorisFlinkUtil.getDataTypeByDorisType(schemaPo.getType()); //类型数组
}

dorisSinkBuilder.setDorisReadOptions(DorisReadOptions.builder().build())
		.setDorisExecutionOptions(executionBuilder.build())
		.setSerializer(
				RowDataSerializer.builder().setFieldNames(fields).setFieldType(types).setType("json").build()
		)
		.setDorisOptions(dorisBuilder.build());

SingleOutputStreamOperator<RowData> mapDs = sourceDs.map(new MapFunction<ConsumerRecord<String, String>, RowData>() {
	@Override
	public RowData map(ConsumerRecord<String, String> record) throws Exception {
		JSONObject json = JSONObject.parseObject(record.value());
		json.put("sampletime", json.getLong("sampletime") + 28800000);
		json.put("plattime", json.getLong("plattime") + 28800000);
		return DorisFlinkUtil.getGenericRowData(json, schemaPoList);
	}
});

mapDs.sinkTo(dorisSinkBuilder.build());

不足：依赖建表语句解析工具及Flink和Doris相关数据类型的转换。搞复杂了。

优化后：

依赖，版本从1.1.0换成了1.0.3

<dependency>
    <groupId>org.apache.doris</groupId>
    <artifactId>flink-doris-connector-1.14_2.12</artifactId>
    <version>1.0.3</version>
</dependency>

代码片段

SingleOutputStreamOperator<String> mapSteam = keyedStream.map(x -> {
	String value = x.value();
	JSONObject jsonObject = JSON.parseObject(value);
	jsonObject.put("sampletime", TimeUtil.getDateStrByFormat(jsonObject.getLong("sampletime")));
	return jsonObject.toJSONString();
});

Properties props = new Properties();
props.setProperty("format", "json");
props.setProperty("strip_outer_array", "true");

SinkFunction<String> dorisSink = DorisSink
		.sink(
				new DorisExecutionOptions.Builder()
						.setBatchIntervalMs(30000L)
						.setBatchSize(1024 * 1024 * 10240)
						.setEnableDelete(false)
						.setMaxRetries(3)
						.setStreamLoadProp(props)
						.build(),
				new DorisOptions.Builder()
						.setFenodes("xx.xx.xx.xx:xxxx")
						.setUsername("xxxx")
						.setPassword("xxxxxx")
						.setTableIdentifier("database.tableName")
						.build()
		);

mapSteam.addSink(dorisSink);

直接将json数据写入Doris，Doris表建议基于json去建，防止字段名称不一致。