flink 使用csv格式写入doris表（UNIQUE KEY） 无法进行部分列更新，使用json可以

[imx7]

使用flink通过flink-connector-doris的rowdata写入doris，flink是1.19.1版本。doris是2.1.6版本。flink-doris-connector 版本是1.19-24.0.0.
建表语句如下，
CREATE TABLE tt.aa (
type int NOT NULL COMMENT '数据来源',
sensor_id varchar(64) NOT NULL COMMENT '车辆id',
latest_time datetime NOT NULL COMMENT '上报时间 yyyy-MM-dd HH:mm:ss',
data_code varchar(64) NULL COMMENT '产品编码',
bike_status int NULL COMMENT '车辆运行',
order_id varchar(64) NULL COMMENT '订单编号',
ws_lat float NULL COMMENT '纬度',
ws_lng float NULL COMMENT '经度',
ws_type int DEFAULT '2' COMMENT '使用的坐标'
)
UNIQUE KEY(type,sensor_id,latest_time)
COMMENT '单车定位信息表'
PARTITION BY RANGE(latest_time)()
DISTRIBUTED BY HASH(type,sensor_id) BUCKETS 4
PROPERTIES (
"replication_allocation" = "tag.location.default: 1",
"dynamic_partition.enable" = "true",
"dynamic_partition.create_history_partition" = "true",
"dynamic_partition.time_unit" = "HOUR",
"dynamic_partition.start" = "-480",
"dynamic_partition.end" = "12",
"dynamic_partition.prefix" = "p"
);
rowdata的配置是

import java.sql.Timestamp;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;
import org.apache.flink.table.data.TimestampData;

public class BPRDMF implements MapFunction<BP, RowData> {

@Override
public RowData map(BP value) throws Exception {
    GenericRowData genericRowData = new GenericRowData(8);
    genericRowData.setField(0, value.type);
    genericRowData.setField(1, StringData.fromString(value.sensorId));
    genericRowData.setField(2, TimestampData.fromTimestamp(Timestamp.valueOf(value.latestTime)));
    genericRowData.setField(3, StringData.fromString(value.dataCode));
    genericRowData.setField(4, value.bikeStatus);
    genericRowData.setField(5, StringData.fromString(value.orderId));
    genericRowData.setField(6, value.wsLat);
    genericRowData.setField(7, value.wsLng);

    return genericRowData;
}

}

flink的doris的sink配置是
String[] fields=new String[] {
"type",
"sensor_id",
"latest_time",
"data_code",
"bike_status",
"order_id",
"ws_lat",
"ws_lng"
};
DataType[] data_types=new DataType[] {
DataTypes.TINYINT(),
DataTypes.VARCHAR(16),
DataTypes.TIMESTAMP(),
DataTypes.VARCHAR(16),
DataTypes.TINYINT(),
DataTypes.VARCHAR(48),
DataTypes.DOUBLE(),
DataTypes.DOUBLE(),
}

public static DorisSink<RowData> generateDorisSink(){
    DorisSink.Builder<RowData> builder = DorisSink.builder();
    DorisOptions.Builder dorisBuilder = DorisOptions.builder();
    dorisBuilder
            .setFenodes("10.1.214.176:8030")
            .setTableIdentifier("tt.aa")
            .setUsername("root")
            .setPassword("123456");

    Properties properties = new Properties();
    
    properties.setProperty("format", "csv");
    properties.setProperty("column_separator", ",");
    properties.setProperty("line_delimiter", "\n");

    properties.setProperty("sink.properties.partial_columns", "true");
    properties.setProperty("sink.properties.columns", "type,sensor_id,latest_time,data_code,bike_status,order_id,ws_lat,ws_lng");

    DorisExecutionOptions.Builder executionBuilder = DorisExecutionOptions.builder();
    executionBuilder
        .setLabelPrefix(table.split("\\.")[1] + "_" + System.currentTimeMillis())
        .setDeletable(false)
        .disable2PC()
        .setBatchMode(true)
        .setBufferFlushMaxRows(10000)
        .setBufferFlushMaxBytes(10 * 1024 * 1024)
        .setBufferFlushIntervalMs(3000)
            .setStreamLoadProp(properties);
    return builder
            .setDorisExecutionOptions(executionBuilder.build())
            .setSerializer(
                    RowDataSerializer
                            .builder()
                            .setFieldNames(fields)
            .setFieldDelimiter(",")
            .setType(LoadConstants.CSV)
                            .setFieldType(data_types)
                            .build())
            .setDorisOptions(dorisBuilder.build())
            .build();
}
positionStream= ...;
positionStream
            .sinkTo(generateDorisSink())
            .name("bike-position-sink");
env.execute("bike-position-info");

Reason: actual column number in csv file is less than schema column number.actual number: 8, schema column number: 9; line delimiter:

如果将csv的配置去掉，修改成JSON格式的，可以正确进行部分列更新。
properties.setProperty("format", "json");
properties.setProperty("read_json_by_line", "true");

RowDataSerializer.builder().setType(LoadConstants.JSON)