Iceberg and Deephaven

Apache Iceberg is a high-performance format for tabular data. Deephaven's Iceberg integration enables users to interact with Iceberg catalogs, namespaces, tables, and snapshots. An Iceberg catalog keeps track of a set of Iceberg tables and groups them into namespaces, much as a database groups its tables into schemas. A snapshot is the state of an Iceberg table at one point in time. Each write to the table creates a new snapshot.

This guide walks through the following tasks:

  • Connect to an Iceberg catalog.
  • Read an Iceberg table into Deephaven as a static table, as a refreshing table, or as of a specific snapshot.
  • Write Deephaven tables to new unpartitioned and partitioned Iceberg tables in the same namespace as the table this guide reads.

The examples in this guide use a REST catalog.

Deephaven's Iceberg integration works with many types of catalogs, including:

  • REST
  • AWS Glue
  • JDBC
  • Hive
  • Hadoop
  • Nessie

Note

Hive and Nessie catalogs require adding their Iceberg catalog modules (iceberg-hive-metastore or iceberg-nessie) to your classpath, and a JDBC catalog requires a JDBC driver for its database.

Deephaven's Iceberg API

The user-facing classes of Deephaven's Iceberg integration are in the io.deephaven.iceberg.util package. The examples in this guide use this package, plus TableDefinition and ColumnDefinition from io.deephaven.engine.table to define table schemas.

A Deephaven deployment for Iceberg

The examples in this guide use the Docker deployment found in the Deephaven example Iceberg REST catalog deployment. It includes a Deephaven server, an Iceberg REST catalog, a RustFS S3-compatible object store, and an Apache Spark server that creates the sample Iceberg table and then shuts down. The repository's README contains a full description of the deployment.

For this guide, you need only clone the repository, cd into the Groovy directory, and run docker compose up:

The deployment creates an Iceberg catalog with one table, taxis, in the nyc namespace. This guide reads that table.

The deployment also uses application mode to define the connection variables that the examples use: catalogUri, awsRegion, awsAccessKeyId, awsSecretAccessKey, and s3Endpoint. Their values come from the deployment's environment variables, so s3Endpoint points to the deployment's RustFS server. If you use a different deployment, define these variables yourself before running the examples.

After the deployment starts, open the Deephaven IDE at http://localhost:10000/ide to run the examples.

Interact with the Iceberg catalog

To interact with an Iceberg catalog, you must first create an IcebergCatalogAdapter instance. Create one with the createAdapter method, which takes a catalog name and the catalog's configuration properties:

The same method creates adapters for the other catalog types listed above. Set the type property to the catalog type, such as "glue" for an AWS Glue catalog, and supply the properties that catalog needs.

Note

If your Iceberg tables are in S3-compatible storage, you can control S3 connectivity with S3Instructions from the io.deephaven.extensions.s3 package. You can pass them in three places:

If you don't pass them, Deephaven derives the S3 settings from the catalog properties. The examples in this guide don't need them.

The catalog adapter can query the namespaces and tables in a catalog. The following code block calls namespaces to get the top-level namespaces and tables to get the tables in the nyc namespace:

Read Iceberg tables

Load an Iceberg table into Deephaven

To load the nyc.taxis Iceberg table, call loadTable. Identify the table as namespace.table, here nyc.taxis. The method returns an IcebergTableAdapter rather than a Deephaven table. The table adapter provides you with several methods to read from or write to the underlying Iceberg table.

To read it into a Deephaven table, pass IcebergReadInstructions to table. This argument is optional. Without it, Deephaven reads the table in static mode. The IcebergReadInstructions reference page lists every instruction you can set.

The following instructions set the update mode to IcebergUpdateMode.staticMode, which loads the table once without refreshing it. Read a refreshing Iceberg table covers the other update modes:

Now that you have the table adapter and the instructions, read the table into a Deephaven table with table:

For greater control over the resultant Deephaven table and greater resilience to schema changes, use an UnboundResolver to map the Iceberg table schema to a Deephaven table definition. The following code maps columns by field ID. Iceberg gives every column in a schema a numeric field ID that stays the same when the column is renamed, so the mapping still works after columns are renamed or reordered. To load the table with the resolver, pass the table identifier and the resolver to loadTable in a LoadTableOptions.

To map a Deephaven column to an Iceberg field by name rather than by field ID, use ColumnInstructions.schemaFieldName, such as putColumnInstructions("PickupTime", ColumnInstructions.schemaFieldName("tpep_pickup_datetime")).

The resolver sets the table definition and column names. IcebergReadInstructions does not set either.

Read a refreshing Iceberg table

Deephaven can also read an Iceberg table as a live, refreshing table. When Deephaven loads a new snapshot of the Iceberg table, the Deephaven table updates, and so does every query built on it. The IcebergUpdateMode class has three update modes:

  • staticMode: Deephaven loads the Iceberg table once and never refreshes it.
  • manualRefreshingMode: Deephaven loads the table once and refreshes it only when you request a refresh.
  • autoRefreshingMode: Deephaven loads the table once and refreshes it at a set interval, which defaults to once per minute.

The examples above use static Iceberg tables. To refresh a manually refreshing table, call update on the IcebergTable that table returns. Without an argument, update refreshes the table to the latest snapshot. To refresh to a specific snapshot instead, pass its ID. The snapshot must be newer than the one the table currently shows. Read a specific snapshot shows how to list snapshot IDs, and Refresh a table after a write shows a manual refresh picking up new data.

To set a refresh interval other than one minute, pass it in milliseconds to autoRefreshingMode.

The following code block sets the update mode in IcebergReadInstructions and reads the same Iceberg table in the other two modes:

Read a specific snapshot

By default, table reads the most recent snapshot. To read a specific snapshot instead, set its ID with the snapshotId method of the IcebergReadInstructions builder.

The snapshots method returns the snapshots of an Iceberg table as a Deephaven table, with each snapshot's ID in the Id column. The listSnapshots method returns them as a list.

The deployment empties the taxis table before loading data into it, so the table's first snapshot has no rows. The following code reads the table as of that first snapshot, so taxisAtSnapshot has the table's columns but no rows:

Write Deephaven tables to Iceberg

Deephaven writes to Iceberg tables by appending data. Each append operation creates a new snapshot. A single append call that writes multiple tables writes them all in the same snapshot. To write data files without committing a snapshot, so that you can commit them in your own Iceberg transaction, use the writer's writeDataFiles method.

To write tables, you create an IcebergTableWriter. A single writer can write any number of Deephaven tables, as long as each table has the writer's table definition without its partitioning columns:

  • For an unpartitioned Iceberg table, each table has the writer's whole table definition.
  • For a partitioned Iceberg table, each table leaves out the partitioning columns, and a partition path supplies their values. See Partitioned Iceberg tables.

Each writer belongs to the Iceberg table whose adapter created it. Creating the writer checks its table definition against the Iceberg table's schema, and writes made with it skip that check, so they are faster. Each write still checks that its tables match the writer's table definition and that its partition paths match the table's partitioning.

Because most of the work happens when you create the writer, create one writer per Iceberg table and reuse it for every Deephaven table you append to that table.

To write Deephaven tables to Iceberg, first create the tables you want to write. This example uses two tables:

Unpartitioned Iceberg tables

When writing data to an unpartitioned Iceberg table, you need the Deephaven table definition:

Then, create the Iceberg table with createTable, which returns an IcebergTableAdapter. It takes a table identifier of the form namespace.table and a table definition. This example creates the table in the nyc namespace, which createTable would create if it didn't exist.

createTable fails if the table already exists. So that you can run this guide more than once, the following code first drops any existing nyc.source table. It calls the dropTable method of the Iceberg catalog that the adapter's catalog method returns. Its second argument, true, also deletes the table's data files. If the table doesn't exist, dropTable returns false instead of failing.

To create a writer, first define the TableParquetWriterOptions that configure it, then pass them to tableWriter:

Now you can write the data to Iceberg. The following code block adds the source2024 and source2025 tables to an IcebergWriteInstructions builder and passes the instructions to append, which writes them to the Iceberg table nyc.source:

Partitioned Iceberg tables

To write data to a partitioned Iceberg table, you must define the TableDefinition yourself and mark one or more columns as partitioning columns. Call withPartitioning on the ColumnDefinition of each partitioning column.

Deephaven can write only to partitioned Iceberg tables whose partition values are plain column values (identity partitioning), with partitioning columns of type boolean, int, long, float, double, String, or LocalDate. For example, it can't append to nyc.taxis, which is partitioned by the day of tpep_pickup_datetime. Tables you create with createTable always use identity partitioning.

The following code defines a table with the partitioning column Year:

Then, create the partitioned Iceberg table with createTable, which returns an IcebergTableAdapter. As before, the code first drops any existing table with the same identifier:

As with unpartitioned tables, create a writer with TableParquetWriterOptions:

Now you can write the data to Iceberg. The addPartitionPaths method of the IcebergWriteInstructions builder takes one partition path per table, in the same order as addTables. A partition path names the partition that a table's rows go to, written as column=value, with pairs separated by / when the table has more than one partitioning column. In this example, Year=2024 sends the rows of source2024 to the 2024 partition.

Drop the partitioning column from each table before writing, because the partition path supplies its value. The following code block uses dropColumns to remove the Year column from the source2024 and source2025 tables and writes them to the nyc.sourcePartitioned table:

Check the write operations

Check that the writes worked by reading each Iceberg table back into Deephaven with the table adapter that created it:

Refresh a table after a write

Now that nyc.source has data, let's watch a manually refreshing table pick up a new snapshot. The following code block reads nyc.source in manual refresh mode and saves a static copy of it with snapshot. It then appends source2025 to nyc.source a second time, which creates a new snapshot, and calls update to refresh sourceManual to that snapshot. sourceManualBefore keeps the original 150 rows, while sourceManual updates to 200 rows:

Any query you build on sourceManual updates along with it. With autoRefreshingMode, Deephaven checks for new snapshots on its own, so your queries keep up with the Iceberg table as other systems write to it.

Next steps

This guide connected Deephaven to an Iceberg catalog, read an Iceberg table as a static table, a refreshing table, and a past snapshot, and appended Deephaven tables to new Iceberg tables. From here, try the examples against your own catalog, or build real-time queries on refreshing Iceberg tables. The reference pages below cover every option.