private String serialize(String tableIdentifier, List values, String format) throws JsonProcessingException {
        if (FORMAT_CSV_VALUE.equalsIgnoreCase(format)) {
            LOG.info("doris data format: {}", format);
            // set columns, and format json data to csv
            String columns = null;
            StringBuilder csvData = new StringBuilder();
            for (Object item : values) {
                if (item instanceof Map) {
                    Map<String, String> map = (Map<String, String>) item;
                    Set<String> fieldNameSet = new LinkedHashSet<>(map.keySet());
                    if (columns == null) {
                        // when single table synchronizing, parameter `sink.properties.columns` in options may
                        // contain hll or bitmap function.
                        // columns: dt,id,name,province,os, pv=hll_hash(id)
                        if (this.columns != null) {
                            for (String fieldName : this.columns.split(",")) {
                                if (fieldName.contains("=")) {
                                    fieldNameSet.add(fieldName);
                                }
                            }
                        }
                        columns = StringUtils.join(fieldNameSet, ",");
                        executionOptions.getStreamLoadProp().put(COLUMNS_KEY, columns);
                    }
                    int idx = 0;
                    int len = map.values().size();
                    for (String val : map.values()) {
                        csvData.append(null == val ? "\\N" : val);
                        if (idx++ < len - 1) {
                            csvData.append(this.fieldDelimiter);
                        }
                    }
                    csvData.append(this.lineDelimiter);
                }
            }
            return csvData.toString();
        } else {
            // Dynamic set COLUMNS_KEY for tableIdentifier every time for multiple sink scenario
            if (multipleSink) {
                executionOptions.getStreamLoadProp().put(COLUMNS_KEY, columnsMap.get(tableIdentifier));
            }
            return OBJECT_MAPPER.writeValueAsString(values);
        }
    }
format data to csv or json

@param values
@param format
@return string
@throws JsonProcessingException
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/doris/src/main/java/org/apache/inlong/sort/doris/table/DorisDynamicSchemaOutputFormat.java