How do I read an Iceberg table from Spark?

Summary

• To read an Iceberg table from Spark, add the Iceberg Spark runtime JAR that matches your Spark and Scala versions, configure a catalog under spark.sql.catalog.<name>, then read the table by name with SQL or spark.table("catalog.namespace.table").
• Time travel uses TIMESTAMP AS OF and VERSION AS OF in SQL, or the snapshot-id and as-of-timestamp read options in the DataFrame API. The as-of-timestamp option takes milliseconds.
• Incremental reads with start-snapshot-id and end-snapshot-id return appended data only, and they aren't available in Spark SQL syntax.
• On Databricks, Unity Catalog managed Iceberg tables are read like any other table by their three-level name, and external Spark can read them through the Iceberg REST catalog endpoint.

How do I read an Iceberg table from Spark?

Add the Iceberg Spark runtime JAR for your Spark and Scala versions, configure a catalog, and read the table by name. The Iceberg docs list 1.12.0 as the latest release, with runtime artifacts such as iceberg-spark-runtime-4.1_2.13 for Spark 4.1 and iceberg-spark-runtime-3.5_2.12 for Spark 3.5. Iceberg uses Spark's DataSourceV2 API, and Spark table identifiers include a catalog name, as in catalog.namespace.table.

from pyspark.sql import SparkSession

spark = (SparkSession.builder
  .config("spark.jars.packages", "org.apache.iceberg:iceberg-spark-runtime-4.1_2.13:1.12.0")
  .config("spark.sql.catalog.prod", "org.apache.iceberg.spark.SparkCatalog")
  .config("spark.sql.catalog.prod.type", "hive")
  .config("spark.sql.catalog.prod.uri", "thrift://metastore-host:port")
  .getOrCreate())

df = spark.table("prod.db.table")
spark.sql("SELECT * FROM prod.db.table").show()

SparkCatalog supports catalogs including Hive, Hadoop, REST, JDBC and Nessie through the type property. SparkSessionCatalog instead adds Iceberg support to Spark's built-in catalog and delegates non-Iceberg tables to it. load() also accepts a path: file:///path/to/table loads a HadoopTable at that path.

Which read syntax fits my goal?

Goal	SQL	DataFrame API
Latest data	SELECT * FROM prod.db.table	spark.table(...)
A snapshot by ID	VERSION AS OF <snapshot-id>	.option("snapshot-id", ...)
State at a time	TIMESTAMP AS OF '<timestamp>'	.option("as-of-timestamp", "<milliseconds>")
A branch or tag	VERSION AS OF '<name>'	.option("branch", ...) or .option("tag", ...)
New appends between snapshots	Not supported	start-snapshot-id and end-snapshot-id options

How do I query an earlier snapshot?

Spark supports TIMESTAMP AS OF and VERSION AS OF in SQL, and FOR SYSTEM_TIME AS OF and FOR SYSTEM_VERSION AS OF also work. A VERSION AS OF value can be a long snapshot ID or a branch or tag name. Find snapshot IDs in the snapshots metadata table.

spark.sql("SELECT * FROM prod.db.table.snapshots")
spark.sql("SELECT * FROM prod.db.table VERSION AS OF 10963874102873")
spark.sql("SELECT * FROM prod.db.table TIMESTAMP AS OF '1986-10-26 01:21:00'")

spark.read.option("snapshot-id", 10963874102873).table("prod.db.table")
spark.read.option("as-of-timestamp", "499162860000").table("prod.db.table")

In SQL, time travel by snapshot ID or timestamp uses that snapshot's schema, so a column added later doesn't appear. Querying a branch uses the table's current schema, and rows written before a column existed return NULL for it.

How do I read only new data between two snapshots?

Set start-snapshot-id (exclusive) and optionally end-snapshot-id (inclusive; the current snapshot if omitted):

spark.read.format("iceberg") \
  .option("start-snapshot-id", "10963874102873") \
  .option("end-snapshot-id", "63874143573109") \
  .load("prod.db.table")

The docs state that this returns only data from append operations. It can't handle replace, overwrite or delete operations, so it isn't a general change feed.

How do schema and partition changes affect reads?

Iceberg schema changes (add, drop, rename, update, reorder) are metadata changes, so no data files are rewritten, and added columns never read values from another column. Hidden partitioning means consumers don't need to know how the table is partitioned or add extra partition filters. After a partition evolution, old data keeps its earlier layout and new data uses the new one. Iceberg uses manifest metadata to prune data files at planning time.

What if I only have the table's location?

Register the existing metadata.json file with the register_table procedure (Iceberg SQL extensions on Spark 3.x, native in Spark 4.0), for example CALL spark_catalog.system.register_table(table => 'db.tbl', metadata_file => 'path/to/metadata/file.json'). The docs warn that registering the same metadata.json in more than one catalog can cause missing updates, data loss and table corruption.

How does Databricks read Iceberg tables?

Unity Catalog managed Iceberg tables need a Unity Catalog workspace, Databricks Runtime 16.4 LTS or above, and serverless compute for managed tables. You read them with ordinary Spark SQL or spark.table using the catalog.schema.table name. Databricks supports Parquet-based Iceberg tables with versions 1, 2 and 3 of the specification, and recommends Iceberg client version 1.9.2 or above for Unity Catalog. Foreign Iceberg tables, managed by catalogs outside Unity Catalog, are read-only in Databricks, and the docs list branching and tagging as unsupported.
To read Unity Catalog tables from your own Spark, set the catalog type to rest and point its uri at the workspace's /api/2.1/unity-catalog/iceberg-rest endpoint. That needs external data access on the metastore and the EXTERNAL USE SCHEMA privilege. The endpoint supports credential vending for some Iceberg reader clients, but external engines can't read views defined in Unity Catalog.

FAQs

Can I use SQL for incremental reads?

No. The Iceberg docs state that incremental read isn't supported by Spark's SQL syntax.

Which JAR do I need for my Spark version?

Use the artifact named for your Spark and Scala versions on the Iceberg multi-engine page, for example iceberg-spark-runtime-3.5_2.13 or iceberg-spark-runtime-4.0_2.13.