Spark Queries🔗

To use Iceberg in Spark, first configureSpark catalogs. Iceberg uses Apache Spark's DataSourceV2 API for data source and catalog implementations.

Querying with SQL🔗

In Spark 3, tables use identifiers that include acatalog name.

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

Metadata tables, likehistory andsnapshots, can use the Iceberg table name as a namespace.

For example, to read from thefiles metadata table forprod.db.table:

SELECT*FROMprod.db.table.files;

Querying with DataFrames🔗

To load a table as a DataFrame, usetable:

valdf=spark.table("prod.db.table")

Catalogs with DataFrameReader🔗

Paths and table names can be loaded with Spark'sDataFrameReader interface. How tables are loaded depends on howthe identifier is specified. When usingspark.read.format("iceberg").load(table) orspark.table(table) thetablevariable can take a number of forms as listed below:

• file:///path/to/table: loads a HadoopTable at given path
• tablename: loadscurrentCatalog.currentNamespace.tablename
• catalog.tablename: loadstablename from the specified catalog.
• namespace.tablename: loadsnamespace.tablename from current catalog
• catalog.namespace.tablename: loadsnamespace.tablename from the specified catalog.
• namespace1.namespace2.tablename: loadsnamespace1.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 usingTIMESTAMP AS OF orVERSION AS OF clauses.TheVERSION AS OF clause can contain a long snapshot ID or a string branch or tag name.

-- time travel to October 26, 1986 at 01:21:00SELECT*FROMprod.db.tableTIMESTAMPASOF'1986-10-26 01:21:00';-- time travel to snapshot with id 10963874102873LSELECT*FROMprod.db.tableVERSIONASOF10963874102873;-- time travel to the head snapshot of audit-branchSELECT*FROMprod.db.tableVERSIONASOF'audit-branch';-- time travel to the snapshot referenced by the tag historical-snapshotSELECT*FROMprod.db.tableVERSIONASOF'historical-snapshot';

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

SELECT*FROMprod.db.tableFORSYSTEM_TIMEASOF'1986-10-26 01:21:00';SELECT*FROMprod.db.tableFORSYSTEM_VERSIONASOF10963874102873;SELECT*FROMprod.db.tableFORSYSTEM_VERSIONASOF'audit-branch';SELECT*FROMprod.db.tableFORSYSTEM_VERSIONASOF'historical-snapshot';

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

-- timestamp in secondsSELECT*FROMprod.db.tableTIMESTAMPASOF499162860;SELECT*FROMprod.db.tableFORSYSTEM_TIMEASOF499162860;

The branch or tag may also be specified using a similar syntax to metadata tables, withbranch_<branchname> ortag_<tagname>:

SELECT*FROMprod.db.table.`branch_audit-branch`;SELECT*FROMprod.db.table.`tag_historical-snapshot`;

(Identifiers with "-" are not valid, and so must be escaped using back quotes.)

Note that the identifier with branch or tag may not be used in combination withVERSION AS OF.

Schema selection in time travel queries🔗

The different time travel queries mentioned in the previous section can use either the snapshot's schema or the table's schema:

-- time travel to October 26, 1986 at 01:21:00 -> uses the snapshot's schemaSELECT*FROMprod.db.tableTIMESTAMPASOF'1986-10-26 01:21:00';-- time travel to snapshot with id 10963874102873L -> uses the snapshot's schemaSELECT*FROMprod.db.tableVERSIONASOF10963874102873;-- time travel to the head of audit-branch -> uses the table's schemaSELECT*FROMprod.db.tableVERSIONASOF'audit-branch';SELECT*FROMprod.db.table.`branch_audit-branch`;-- time travel to the snapshot referenced by the tag historical-snapshot -> uses the snapshot's schemaSELECT*FROMprod.db.tableVERSIONASOF'historical-snapshot';SELECT*FROMprod.db.table.`tag_historical-snapshot`;

DataFrame🔗

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

• snapshot-id selects a specific table snapshot
• as-of-timestamp selects the current snapshot at a timestamp, in milliseconds
• branch selects the head snapshot of the specified branch. Note that currently branch cannot be combined with as-of-timestamp.
• tag selects the snapshot associated with the specified tag. Tags cannot be combined withas-of-timestamp.

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

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

// time travel to tag historical-snapshotspark.read.option(SparkReadOptions.TAG,"historical-snapshot").format("iceberg").load("path/to/table")

// time travel to the head snapshot of audit-branchspark.read.option(SparkReadOptions.BRANCH,"audit-branch").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")

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 fordb.table is read usingdb.table.history.

History🔗

To show table history:

SELECT*FROMprod.db.table.history;

Metadata Log Entries🔗

To show table metadata log entries:

SELECT*fromprod.db.table.metadata_log_entries;

Snapshots🔗

To show the valid snapshots for a table:

SELECT*FROMprod.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:

selecth.made_current_at,s.operation,h.snapshot_id,h.is_current_ancestor,s.summary['spark.app.id']fromprod.db.table.historyhjoinprod.db.table.snapshotssonh.snapshot_id=s.snapshot_idorderbymade_current_at;

Entries🔗

To show all the table's current manifest entries for both data and delete files.

SELECT*FROMprod.db.table.entries;

Files🔗

To show a table's current files:

SELECT*FROMprod.db.table.files;

To show only data files or delete files, queryprod.db.table.data_files andprod.db.table.delete_files respectively.To show all files, data files and delete files across all tracked snapshots, queryprod.db.table.all_files,prod.db.table.all_data_files andprod.db.table.all_delete_files respectively.

Manifests🔗

To show a table's current file manifests:

SELECT*FROMprod.db.table.manifests;

Note:

1. Fields withinpartition_summaries column of the manifests table correspond tofield_summary structs withinmanifest 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, wherecontains_nan is not populated.

Partitions🔗

To show a table's current partitions:

SELECT*FROMprod.db.table.partitions;

Note:

1. For unpartitioned tables, the partitions table will not contain the partition and spec_id fields.

2. The partitions metadata table shows partitions with data files or delete files in the current snapshot. However, delete files are not applied, and so in some cases partitions may be shown even though all their data rows are marked deleted by delete files.

Positional Delete Files🔗

To show all positional delete files from the current snapshot of table:

SELECT*fromprod.db.table.position_deletes;

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*FROMprod.db.table.all_data_files;

All Delete Files🔗

To show the table's delete files and each file's metadata from all the snapshots:

SELECT*FROMprod.db.table.all_delete_files;

All Entries🔗

To show the table's manifest entries from all the snapshots for both data and delete files:

SELECT*FROMprod.db.table.all_entries;

All Manifests🔗

To show all of the table's manifest files:

SELECT*FROMprod.db.table.all_manifests;

Note:

1. Fields withinpartition_summaries column of the manifests table correspond tofield_summary structs withinmanifest 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, wherecontains_nan is not populated.

References🔗

To show a table's known snapshot references:

SELECT*FROMprod.db.table.refs;

Inspecting with DataFrames🔗

Metadata tables can be loaded using the DataFrameReader API:

// named metastore tablespark.read.format("iceberg").load("db.table.files")// Hadoop path tablespark.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:00SELECT*FROMprod.db.table.manifestsTIMESTAMPASOF'2021-09-20 08:00:00';-- get the table's partitions with snapshot id 10963874102873LSELECT*FROMprod.db.table.partitionsVERSIONASOF10963874102873;

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

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