public static DataFile writeFile(
            Table table,
            Schema schema,
            PartitionSpec spec,
            Configuration conf,
            String location,
            String filename,
            List<RowData> rows,
            StructLike partition)
            throws IOException {
        Path path = new Path(location, filename);
        FileFormat fileFormat = FileFormat.fromFileName(filename);
        Preconditions.checkNotNull(fileFormat, "Cannot determine format for file: %s", filename);

        RowType flinkSchema = FlinkSchemaUtil.convert(schema);
        FileAppenderFactory<RowData> appenderFactory =
                new FlinkAppenderFactory(
                        schema, flinkSchema, ImmutableMap.of(), spec, null, null, null);

        FileAppender<RowData> appender = appenderFactory.newAppender(fromPath(path, conf), fileFormat);
        try (FileAppender<RowData> closeableAppender = appender) {
            closeableAppender.addAll(rows);
        }

        DataFiles.Builder builder =
                DataFiles.builder(spec)
                        .withInputFile(HadoopInputFile.fromPath(path, conf))
                        .withMetrics(appender.metrics());

        if (partition != null) {
            builder = builder.withPartition(partition);
        }

        return builder.build();
    }
Write the list of {@link RowData} to the given path and with the given partition data
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/test/java/org/apache/inlong/sort/iceberg/sink/util/SimpleDataUtil.java