@Override
    public void write(IN element, long currentTime) {
        String databaseName = null;
        String tableName = null;
        int recordNum = 1;
        int recordSize = 0;
        JsonNode rootNode = null;
        ObjectIdentifier identifier = null;
        HashMap<ObjectIdentifier, Long> ignoreWritingTableMap = CacheHolder.getIgnoreWritingTableMap();
        try {
            if (sinkMultipleEnable) {
                GenericRowData rowData = (GenericRowData) element;
                if (jsonFormat == null) {
                    jsonFormat = (JsonDynamicSchemaFormat) DynamicSchemaFormatFactory.getFormat(sinkMultipleFormat);
                }
                byte[] rawData = (byte[]) rowData.getField(0);
                rootNode = jsonFormat.deserialize(rawData);
                LOG.debug("root node: {}", rootNode);
                boolean isDDL = jsonFormat.extractDDLFlag(rootNode);
                if (isDDL) {
                    // Ignore ddl change for now
                    return;
                }
                databaseName = jsonFormat.parse(rootNode, databasePattern);
                tableName = jsonFormat.parse(rootNode, tablePattern);

                List<Map<String, Object>> physicalDataList = HiveTableUtil.jsonNode2Map(
                        jsonFormat.getPhysicalData(rootNode));
                recordNum = physicalDataList.size();

                identifier = HiveTableUtil.createObjectIdentifier(databaseName, tableName);

                // ignore writing data into this table
                if (ignoreWritingTableMap.containsKey(identifier)) {
                    return;
                }

                List<String> pkListStr = jsonFormat.extractPrimaryKeyNames(rootNode);
                RowType schema = jsonFormat.extractSchema(rootNode, pkListStr);
                HiveWriterFactory writerFactory = getHiveWriterFactory(identifier, schema, hiveVersion);

                // parse the real location of hive table
                Path inProgressFilePath = getInProgressPath(writerFactory);

                LOG.debug("in progress file path: {}", inProgressFilePath);
                Pair<RecordWriter, Function<RowData, Writable>> pair = getRecordWriterAndRowConverter(writerFactory,
                        inProgressFilePath);
                // reset record writer and row converter
                writer.setRecordWriter(pair.getLeft());
                writer.setRowConverter(pair.getRight());
                writer.setInProgressPath(inProgressFilePath);

                boolean replaceLineBreak = writerFactory.getStorageDescriptor().getInputFormat()
                        .contains("TextInputFormat");

                for (Map<String, Object> record : physicalDataList) {
                    // check and alter hive table if schema has changed
                    boolean changed = checkSchema(identifier, writerFactory, schema);
                    if (changed) {
                        // remove cache and reload hive writer factory
                        CacheHolder.getFactoryMap().remove(identifier);
                        writerFactory = HiveTableUtil.getWriterFactory(hiveShim, hiveVersion, identifier);
                        assert writerFactory != null;
                        FileSinkOperator.RecordWriter recordWriter = writerFactory.createRecordWriter(
                                inProgressFilePath);
                        Function<RowData, Writable> rowConverter = writerFactory.createRowDataConverter();
                        writer.setRecordWriter(recordWriter);
                        writer.setRowConverter(rowConverter);
                        CacheHolder.getRecordWriterHashMap().put(inProgressFilePath, recordWriter);
                        CacheHolder.getRowConverterHashMap().put(inProgressFilePath, rowConverter);
                    }

                    LOG.debug("record: {}", record);
                    LOG.debug("columns : {}", Arrays.deepToString(writerFactory.getAllColumns()));
                    LOG.debug("types: {}", Arrays.deepToString(writerFactory.getAllTypes()));
                    Pair<GenericRowData, Integer> rowDataPair = HiveTableUtil.getRowData(record,
                            writerFactory.getAllColumns(), writerFactory.getAllTypes(), replaceLineBreak);
                    GenericRowData genericRowData = rowDataPair.getLeft();
                    recordSize += rowDataPair.getRight();
                    LOG.debug("generic row data: {}", genericRowData);
                    writer.addElement(genericRowData);
                }
                if (metricData != null) {
                    metricData.outputMetrics(databaseName, tableName, recordNum, recordSize);
                }
            } else {
                RowData data = (RowData) element;
                writer.addElement(data);
                if (metricData != null) {
                    if (data instanceof BinaryRowData) {
                        // mysql cdc sends BinaryRowData
                        metricData.invoke(1, ((BinaryRowData) data).getSizeInBytes());
                    } else {
                        // oracle cdc sends GenericRowData
                        metricData.invoke(1, data.toString().getBytes(StandardCharsets.UTF_8).length);
                    }
                }
            }
        } catch (Exception e) {
            if (schemaUpdatePolicy == null || SchemaUpdateExceptionPolicy.THROW_WITH_STOP == schemaUpdatePolicy) {
                throw new FlinkHiveException("Failed to write data", e);
            } else if (SchemaUpdateExceptionPolicy.STOP_PARTIAL == schemaUpdatePolicy) {
                if (identifier != null) {
                    ignoreWritingTableMap.put(identifier, 1L);
                }
            } else if (SchemaUpdateExceptionPolicy.LOG_WITH_IGNORE == schemaUpdatePolicy) {
                handleDirtyData(databaseName, tableName, recordNum, recordSize, rootNode, jsonFormat, e);
            }
            LOG.error("Failed to write data", e);
        }
        markWrite(currentTime);
    }
The part-file writer that writes to the specified hadoop path.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/hive/src/main/java/org/apache/inlong/sort/hive/filesystem/HadoopPathBasedPartFileWriter.java