private DataStreamSink<Row> createBatchSink(
            DataStream<RowData> dataStream,
            DataStructureConverter converter,
            StorageDescriptor sd,
            HiveWriterFactory recordWriterFactory,
            OutputFileConfig fileNaming,
            final int parallelism)
            throws IOException {
        FileSystemOutputFormat.Builder<Row> builder = new FileSystemOutputFormat.Builder<>();
        builder.setPartitionComputer(
                new HiveRowPartitionComputer(
                        hiveShim,
                        JobConfUtils.getDefaultPartitionName(jobConf),
                        tableSchema.getFieldNames(),
                        tableSchema.getFieldDataTypes(),
                        getPartitionKeyArray()));
        builder.setDynamicGrouped(dynamicGrouping);
        builder.setPartitionColumns(getPartitionKeyArray());
        builder.setFileSystemFactory(fsFactory());
        builder.setFormatFactory(new HiveOutputFormatFactory(recordWriterFactory));
        builder.setMetaStoreFactory(msFactory());
        builder.setOverwrite(overwrite);
        builder.setStaticPartitions(staticPartitionSpec);
        builder.setTempPath(
                new org.apache.flink.core.fs.Path(toStagingDir(sd.getLocation(), jobConf)));
        builder.setOutputFileConfig(fileNaming);
        return dataStream
                .map((MapFunction<RowData, Row>) value -> (Row) converter.toExternal(value))
                .writeUsingOutputFormat(builder.build())
                .setParallelism(parallelism);
    }
Table sink to write to Hive tables.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/hive/src/main/java/org/apache/inlong/sort/hive/HiveTableSink.java