@Override
    public void open(int taskNumber, int numTasks) throws IOException {
        Properties loadProps = executionOptions.getStreamLoadProp();
        dorisStreamLoad = new DorisStreamLoad(getBackend(), options.getUsername(), options.getPassword(), loadProps);
        if (!multipleSink) {
            this.jsonFormat = true;
            // handleStreamLoadProp();
            this.fieldGetters = new RowData.FieldGetter[logicalTypes.length];
            for (int i = 0; i < logicalTypes.length; i++) {
                fieldGetters[i] = DorisParseUtils.createFieldGetter(logicalTypes[i], i);
            }
            try {
                schema = RestService.getSchema(options, readOptions, LOG);
            } catch (DorisException e) {
                throw new RuntimeException(e);
            }
        }

        if (multipleSink) {
            jsonDynamicSchemaFormat =
                    (JsonDynamicSchemaFormat) DynamicSchemaFormatFactory.getFormat(dynamicSchemaFormat);
            helper = SchemaChangeHelper.of(jsonDynamicSchemaFormat, options, enableSchemaChange,
                    enableSchemaChange ? SchemaChangeUtils.deserialize(schemaChangePolicies) : null, databasePattern,
                    tablePattern, executionOptions.getMaxRetries(), schemaUpdatePolicy, metricData, dirtySinkHelper);
        }
        MetricOption metricOption = MetricOption.builder()
                .withInlongLabels(inlongMetric)
                .withAuditAddress(auditHostAndPorts)
                .withInitRecords(metricState != null ? metricState.getMetricValue(NUM_RECORDS_OUT) : 0L)
                .withInitBytes(metricState != null ? metricState.getMetricValue(NUM_BYTES_OUT) : 0L)
                .withInitDirtyRecords(metricState != null ? metricState.getMetricValue(DIRTY_RECORDS_OUT) : 0L)
                .withInitDirtyBytes(metricState != null ? metricState.getMetricValue(DIRTY_BYTES_OUT) : 0L)
                .withRegisterMetric(MetricOption.RegisteredMetric.ALL)
                .build();
        if (metricOption != null) {
            metricData = new SinkTableMetricData(metricOption, getRuntimeContext().getMetricGroup());
            if (multipleSink) {
                metricData.registerSubMetricsGroup(metricState);
            }
        }
        dirtySinkHelper.open(new Configuration());
        if (executionOptions.getBatchIntervalMs() != 0 && executionOptions.getBatchSize() != 1) {
            this.scheduler = new ScheduledThreadPoolExecutor(1,
                    new ExecutorThreadFactory("doris-streamload-output-format"));
            this.scheduledFuture = this.scheduler.scheduleWithFixedDelay(() -> {
                if (!closed && !flushing) {
                    flush();
                }
            }, executionOptions.getBatchIntervalMs(), executionOptions.getBatchIntervalMs(), TimeUnit.MILLISECONDS);
        }
    }
A builder used to set parameters to the output format's configuration in a fluent way.

@return builder
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/doris/src/main/java/org/apache/inlong/sort/doris/table/DorisDynamicSchemaOutputFormat.java