Spark Queries

To use Iceberg in Spark, first configure Spark catalogs.

Iceberg uses Apache Spark’s DataSourceV2 API for data source and catalog implementations. Spark DSv2 is an evolving API with different levels of support in Spark versions:

Querying with SQL

In Spark 3, tables use identifiers that include a catalog name.

SELECT * FROM prod.db.table; -- catalog: prod, namespace: db, table: table

Metadata tables, like history and snapshots, can use the Iceberg table name as a namespace.

For example, to read from the files metadata table for prod.db.table:

SELECT * FROM prod.db.table.files;

Querying with DataFrames

To load a table as a DataFrame, use table:

val df = spark.table("prod.db.table")

Catalogs with DataFrameReader

Iceberg 0.11.0 adds multi-catalog support to DataFrameReader in both Spark 3.x and 2.4.

Paths and table names can be loaded with Spark’s DataFrameReader interface. How tables are loaded depends on how the identifier is specified. When using spark.read.format("iceberg").load(table) or spark.table(table) the table variable can take a number of forms as listed below:

• file:///path/to/table: loads a HadoopTable at given path
• tablename: loads currentCatalog.currentNamespace.tablename
• catalog.tablename: loads tablename from the specified catalog.
• namespace.tablename: loads namespace.tablename from current catalog
• catalog.namespace.tablename: loads namespace.tablename from the specified catalog.
• namespace1.namespace2.tablename: loads namespace1.namespace2.tablename from current catalog

The above list is in order of priority. For example: a matching catalog will take priority over any namespace resolution.

Time travel

SQL

Spark 3.3 and later supports time travel in SQL queries using TIMESTAMP AS OF or VERSION AS OF clauses

-- time travel to October 26, 1986 at 01:21:00
SELECT * FROM prod.db.table TIMESTAMP AS OF '1986-10-26 01:21:00';

-- time travel to snapshot with id 10963874102873L
SELECT * FROM prod.db.table VERSION AS OF 10963874102873;

In addition, FOR SYSTEM_TIME AS OF and FOR SYSTEM_VERSION AS OF clauses are also supported:

SELECT * FROM prod.db.table FOR SYSTEM_TIME AS OF '1986-10-26 01:21:00';
SELECT * FROM prod.db.table FOR SYSTEM_VERSION AS OF 10963874102873;

Timestamps may also be supplied as a Unix timestamp, in seconds:

-- timestamp in seconds
SELECT * FROM prod.db.table TIMESTAMP AS OF 499162860;
SELECT * FROM prod.db.table FOR SYSTEM_TIME AS OF 499162860;

DataFrame

To select a specific table snapshot or the snapshot at some time in the DataFrame API, Iceberg supports two Spark read options:

• snapshot-id selects a specific table snapshot
• as-of-timestamp selects the current snapshot at a timestamp, in milliseconds

// time travel to October 26, 1986 at 01:21:00
spark.read
    .option("as-of-timestamp", "499162860000")
    .format("iceberg")
    .load("path/to/table")

// time travel to snapshot with ID 10963874102873L
spark.read
    .option("snapshot-id", 10963874102873L)
    .format("iceberg")
    .load("path/to/table")

Incremental read

To read appended data incrementally, use:

• start-snapshot-id Start snapshot ID used in incremental scans (exclusive).
• end-snapshot-id End snapshot ID used in incremental scans (inclusive). This is optional. Omitting it will default to the current snapshot.

// get the data added after start-snapshot-id (10963874102873L) until end-snapshot-id (63874143573109L)
spark.read()
  .format("iceberg")
  .option("start-snapshot-id", "10963874102873")
  .option("end-snapshot-id", "63874143573109")
  .load("path/to/table")

Spark 2.4

Spark 2.4 requires using the DataFrame reader with iceberg as a format, because 2.4 does not support direct SQL queries:

// named metastore table
spark.read.format("iceberg").load("catalog.db.table")
// Hadoop path table
spark.read.format("iceberg").load("hdfs://nn:8020/path/to/table")

Spark 2.4 with SQL

To run SQL SELECT statements on Iceberg tables in 2.4, register the DataFrame as a temporary table:

val df = spark.read.format("iceberg").load("db.table")
df.createOrReplaceTempView("table")

spark.sql("""select count(1) from table""").show()

Inspecting tables

To inspect a table’s history, snapshots, and other metadata, Iceberg supports metadata tables.

Metadata tables are identified by adding the metadata table name after the original table name. For example, history for db.table is read using db.table.history.

History

To show table history:

SELECT * FROM prod.db.table.history;

Metadata Log Entries

To show table metadata log entries:

SELECT * from prod.db.table.metadata_log_entries;

Snapshots

To show the valid snapshots for a table:

SELECT * FROM prod.db.table.snapshots;

You can also join snapshots to table history. For example, this query will show table history, with the application ID that wrote each snapshot:

select
    h.made_current_at,
    s.operation,
    h.snapshot_id,
    h.is_current_ancestor,
    s.summary['spark.app.id']
from prod.db.table.history h
join prod.db.table.snapshots s
  on h.snapshot_id = s.snapshot_id
order by made_current_at

Files

To show a table’s current data files:

SELECT * FROM prod.db.table.files;

Manifests

To show a table’s current file manifests:

SELECT * FROM prod.db.table.manifests;

Note:

1. Fields within partition_summaries column of the manifests table correspond to field_summary structs within manifest list, with the following order:
   • contains_null
   • contains_nan
   • lower_bound
   • upper_bound
2. contains_nan could return null, which indicates that this information is not available from the file’s metadata. This usually occurs when reading from V1 table, where contains_nan is not populated.

Partitions

To show a table’s current partitions:

SELECT * FROM prod.db.table.partitions;

Note: For unpartitioned tables, the partitions table will contain only the record_count and file_count columns.

All Metadata Tables

These tables are unions of the metadata tables specific to the current snapshot, and return metadata across all snapshots.

All Data Files

To show all of the table’s data files and each file’s metadata:

SELECT * FROM prod.db.table.all_data_files;

All Manifests

To show all of the table’s manifest files:

SELECT * FROM prod.db.table.all_manifests;

Note:

1. Fields within partition_summaries column of the manifests table correspond to field_summary structs within manifest list, with the following order:
   • contains_null
   • contains_nan
   • lower_bound
   • upper_bound
2. contains_nan could return null, which indicates that this information is not available from the file’s metadata. This usually occurs when reading from V1 table, where contains_nan is not populated.

References

To show a table’s known snapshot references:

SELECT * FROM prod.db.table.refs;

Inspecting with DataFrames

Metadata tables can be loaded in Spark 2.4 or Spark 3 using the DataFrameReader API:

// named metastore table
spark.read.format("iceberg").load("db.table.files")
// Hadoop path table
spark.read.format("iceberg").load("hdfs://nn:8020/path/to/table#files")

Time Travel with Metadata Tables

To inspect a tables’s metadata with the time travel feature:

-- get the table's file manifests at timestamp Sep 20, 2021 08:00:00
SELECT * FROM prod.db.table.manifests TIMESTAMP AS OF '2021-09-20 08:00:00';

-- get the table's partitions with snapshot id 10963874102873L
SELECT * FROM prod.db.table.partitions VERSION AS OF 10963874102873;

Metadata tables can also be inspected with time travel using the DataFrameReader API:

// load the table's file metadata at snapshot-id 10963874102873 as DataFrame
spark.read.format("iceberg").option("snapshot-id", 10963874102873L).load("db.table.files")