Netflix Iceberg

Iceberg is organized into several library modules depending on your integration needs:

Core Library Modules:

• iceberg-api: Contains the public Iceberg API.
• iceberg-core: The primary implementation of the Iceberg API and support for Avro data files. Processing engines should depend on this module.
• iceberg-common: Utility classes used across other modules.
• iceberg-parquet: Optional module for tables backed by Parquet files.
• iceberg-orc: Optional module for tables backed by ORC files (experimental).
• iceberg-hive: Implementation for tables backed by the Hive Metastore Thrift client.

Engine Integration Modules:

• iceberg-spark: Implementation of Spark's Datasource V2 API. Use iceberg-runtime for a shaded version.
• iceberg-presto-runtime: A shaded runtime jar for Presto integration.
• iceberg-pig: Implementation of Pig's LoadFunc API.
• iceberg-data: A client library for reading Iceberg tables from JVM applications.

Iceberg is a table format for large, slow-moving tabular data that tracks individual data files instead of relying on directory structures.

Core Concepts:

• Metadata Files: Maintain the table state, including schema, partitioning configuration, properties, and a list of snapshots. All changes to the table state are performed via atomic operations that replace the old metadata file with a new one.
• Snapshots: Represent the complete set of data files in a table at a specific point in time. This provides snapshot isolation, ensuring readers see a consistent view of the data without locks.
• Manifest Files: Store the actual data file locations, partition data, and metrics. Snapshots are composed of the union of files found in their associated manifest files.
• Atomic Commits: Because changes are metadata-only operations, Iceberg enables safe file-level operations like compaction and appending late data without the need for expensive file renames (which are problematic on object stores like S3).

In Iceberg, GenericDataFile is a concrete implementation of the DataFile interface. It serves as a metadata object representing a single data file within a table. It implements several interfaces including IndexedRecord and StructLike, allowing it to be used with Avro reflection and other structured data processing frameworks.

Key properties captured by a GenericDataFile include:

• File Identity: path() (the file path) and format() (the FileFormat).
• Partitioning: partition() (returns a StructLike representing the partition values).
• Statistics: recordCount(), fileSizeInBytes(), blockSizeInBytes(), and various column-level metrics like columnSizes(), valueCounts(), nullValueCounts(), lowerBounds(), and upperBounds().
• Ordering: sortColumns() and fileOrdinal().

Because it implements IndexedRecord, it can be instantiated via Avro reflection using an org.apache.avro.Schema constructor, which is particularly useful when reading manifest files.

To add existing data files to an Iceberg table, use the newFastAppend API. This process typically involves:

1. Listing partition files using SparkTableUtil.listPartition.
2. Loading the table via HadoopTables.
3. Using table.newFastAppend to create an append operation.
4. Calling appendFile(file.toDataFile(table.spec)) for each file.
5. Calling commit() to finalize the changes.

For a single large commit (creating one manifest instead of many), consider using table.newAppend.commit.

import com.netflix.iceberg.hadoop.HadoopTables
import org.apache.hadoop.conf.Configuration

// Assuming 'partitions' DataFrame is available from SparkTableUtil.partitionDF
partitions.repartition(100).flatMap { row =>

    // List the partition and read Parquet footers to get metrics
    SparkTableUtil.listPartition(row.getMap[String, String](0).toMap, row.getString(1), row.getString(2))

}.repartition(10) 
 .mapPartitions { files =>

    val tables = new HadoopTables(new Configuration())
    val table = tables.load("hdfs:/tmp/tables/job_metrics_tmp")

    // Use fast appends to create a manifest for the new files
    val append = table.newFastAppend

    files.foreach { file =>
        append.appendFile(file.toDataFile(table.spec))
    }

    // Commit the new files
    append.commit()

    Seq.empty[String].iterator

}.count

You can create an Iceberg table in HDFS using HadoopTables and SparkSchemaUtil. SparkSchemaUtil allows you to derive the schema and partition specification directly from an existing Spark table, which is useful for migrating data to Iceberg.

Note: When using Spark objects inside transformations or closures, use a code block to avoid capturing Spark configurations in closures.

import org.apache.hadoop.fs.Path
import com.netflix.iceberg.hadoop.HadoopTables
import com.netflix.iceberg.spark.SparkSchemaUtil

val path = "hdfs:/tmp/tables/job_metrics_tmp"

{
    val conf = spark.sparkContext.hadoopConfiguration
    val fs = new Path(path).getFileSystem(conf)
    fs.delete(new Path(path), true /* recursive */ )

    val tables = new HadoopTables(conf)
    // Derive schema and spec from an existing Spark table
    val schema = SparkSchemaUtil.schemaForTable(spark, "default.job_metrics")
    val spec = SparkSchemaUtil.specForTable(spark, "default.job_metrics")

    tables.create(schema, spec, path)

    // Verify the schema
    tables.load(path).schema
}