Iceberg表无法识别生成的Parquet文件问题求助

问题核心 #

你手动生成Parquet文件后仅将其放在表目录下，但Iceberg依赖自身元数据（快照、清单文件）追踪数据文件，未注册的文件不会被Java API识别；而Python用pyarrow直接读取Parquet文件，不依赖Iceberg元数据，所以能正常读取。

解决步骤 #

要让Iceberg识别生成的Parquet文件，需将文件的元数据信息（路径、分区、记录数、大小等）添加到Iceberg表的快照中，可通过Transaction实现，以下是基于你现有代码的修改方案：

修改后的完整代码 #

import org.apache.hadoop.conf.Configuration;
import org.apache.iceberg.*;
import org.apache.iceberg.catalog.Catalog;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.data.GenericRecord;
import org.apache.iceberg.data.parquet.GenericParquetWriter;
import org.apache.iceberg.hadoop.HadoopCatalog;
import org.apache.iceberg.io.FileAppender;
import org.apache.iceberg.parquet.Parquet;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.types.Types;

import java.io.File;
import java.io.IOException;
import java.time.LocalDate;
import java.time.temporal.ChronoUnit;
import java.util.List;

import static org.apache.iceberg.types.Types.NestedField.optional;
import static org.apache.iceberg.types.Types.NestedField.required;

public class IcebergTableAppend {
    public static void main(String[] args) {
        System.out.println("Appending records ");
        Configuration conf = new Configuration();
        String lakehouse = "/tmp/iceberg-test";
        conf.set(CatalogProperties.WAREHOUSE_LOCATION, lakehouse);
        Schema schema = new Schema(
                required(1, "hotel_id", Types.LongType.get()),
                optional(2, "hotel_name", Types.StringType.get()),
                required(3, "customer_id", Types.LongType.get()),
                required(4, "arrival_date", Types.DateType.get()),
                required(5, "departure_date", Types.DateType.get()),
                required(6, "value", Types.DoubleType.get())
        );
        PartitionSpec spec = PartitionSpec.builderFor(schema)
                .month("arrival_date")
                .build();
        TableIdentifier id = TableIdentifier.parse("bookings.rome_hotels");
        String warehousePath = "file://" + lakehouse;
        Catalog catalog = new HadoopCatalog(conf, warehousePath);
        // rm -rf  /tmp/iceberg-test/bookings
        Table table = catalog.createTable(id, schema, spec);
        List<GenericRecord> records = Lists.newArrayList();
        // generating a bunch of records
        for (int j = 1; j <= 12; j++) {
            int NUM_ROWS_PER_MONTH = 2300;
            for (int i = 0; i < NUM_ROWS_PER_MONTH; i++) {
                GenericRecord rec = GenericRecord.create(schema);
                rec.setField("hotel_id", (long) (i * 2) + 10000);
                rec.setField("hotel_name", "hotel_name-" + i + 1000);
                rec.setField("customer_id", (long) (i * 2) + 20000);
                rec.setField("arrival_date",
                        LocalDate.of(2022, j, (i % 23) + 1)
                                .plus(1, ChronoUnit.DAYS));
                rec.setField("departure_date",
                        LocalDate.of(2022, j, (i % 23) + 5));
                rec.setField("value", (double) i * 4.13);
                records.add(rec);
            }
        }
        File parquetFile = new File(
                lakehouse + "/bookings/rome_hotels/arq_001.parquet");
        FileAppender<GenericRecord> appender = null;
        try {
            appender = Parquet.write(Files.localOutput(parquetFile))
                    .schema(table.schema())
                    .createWriterFunc(GenericParquetWriter::buildWriter)
                    .build();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        try {
            appender.addAll(records);
        } finally {
            try {
                appender.close();
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }

        // 关键：将生成的Parquet文件关联到Iceberg表元数据
        try {
            // 1. 转换为Iceberg兼容的路径格式
            org.apache.iceberg.Path icebergFilePath = new org.apache.iceberg.Path(parquetFile.toURI().toString());
            
            // 2. 获取文件元数据：大小、记录数
            long fileSize = parquetFile.length();
            long recordCount = Parquet.readMetadata(Files.localInput(parquetFile)).rowCount();
            
            // 3. 构建DataFile对象（示例简化处理，实际跨分区需按分区拆分）
            DataFile dataFile = DataFiles.builder(spec)
                    .format(FileFormat.PARQUET)
                    .path(icebergFilePath)
                    .fileSizeInBytes(fileSize)
                    .recordCount(recordCount)
                    .build();
            
            // 4. 提交事务，将DataFile添加到表快照
            Transaction transaction = table.newTransaction();
            transaction.newAppend().appendFile(dataFile).commit();
            transaction.commit();
            
            System.out.println("文件已成功关联到Iceberg表");
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}

关键代码说明 #

• DataFile构建：通过DataFiles.builder生成文件元数据对象，必须指定文件格式、路径、大小、记录数；如果是分区表，需额外指定分区信息（示例因数据跨12个月做了简化，生产环境需按分区拆分生成对应DataFile）。
• 事务提交：通过table.newTransaction()开启原子事务，调用newAppend()添加DataFile后提交，确保Iceberg元数据的一致性。

注意事项 #

1. 若数据跨多个分区，需按分区分组统计记录数和文件片段，生成对应分区的DataFile，否则无法利用Iceberg的分区过滤能力。
2. 生产环境建议优先使用Iceberg原生的AppendFiles或FileWriter API写入数据，无需手动处理Parquet文件和元数据关联。
3. 确保Iceberg 1.0.0依赖包完整，避免API兼容问题。

内容的提问来源于stack exchange，提问作者João Paraná