Iceberg 表看不到生成的 Parquet 文件

Question

在我的用例中，创建了 Iceberg 格式的表。它只接收 APPEND 操作，因为它是关于在时间序列流中记录事件。评估使用情况冰山格式在这个用例中，我创建了一个简单的 Java 程序，它创建了一组 27600 行。元数据和 parquet 文件都已创建，但我无法通过 Java API (https://iceberg.apache.org/docs/latest/java-api-quickstart/) 访问它们。我正在使用HadoopCatalog 和FileAppender<GenericRecord>。重要的是，我可以通过 Python 3 脚本读取使用 pyarrow 和 datafusion 模块创建的 Parquet 文件，这是正确的！

我相信我的程序中将生成的 Parquet 文件链接到目录中创建的表的某些方法的执行必须丢失。

注意：我只在 1.0.0 版本中使用 Apache Iceberg 的 Java API

API 中有一个 org.apache.iceberg.Transaction 对象，它接受 org.apache.iceberg.DataFile 但我还没有看到如何使用它的示例，我也不知道它是否对解决这个问题有用.

请参阅下面的程序：

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);
            }
        }
    }
}

Answer 1

我发现了如何修复 Java 程序。

只需将以下几行添加到 main 方法的末尾

PartitionKey partitionKey = new PartitionKey(table.spec(), table.schema());
DataFile dataFile = DataFiles.builder(table.spec())
        .withPartition(partitionKey)
        .withInputFile(localInput(parquetFile))
        .withMetrics(appender.metrics())
        .withFormat(FileFormat.PARQUET)
        .build();
 Transaction t = table.newTransaction();
 t.newAppend().appendFile(dataFile).commit();
 // commit all changes to the table
 t.commitTransaction();

还将以下依赖项添加到您的 POM 文件中

<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-mapreduce-client-core</artifactId>
    <version>3.3.4</version>
</dependency>

这避免了如下所示的运行时错误：

java.lang.ClassNotFoundException: org.apache.hadoop.mapreduce.lib.input.FileInputFormat