Apache Iceberg and time travel queries with Scala/Spark

Iceberg provides time travel functionality, allowing you to query the state of a table at a previous point in time. You can query the table as it existed at a specific snapshot or as of a certain timestamp.

Example: querying an Iceberg table at a specific snapshot

Each time you make changes (like insert or update) to an Iceberg table, a snapshot is created. You can query data from a particular snapshot.

def timeTravelBySnapshot(spark: SparkSession, snapshotId: Long): Unit = {
  val timeTravelDF = spark.read
    .option("snapshot-id", snapshotId)
    .table("spark_catalog.default.iceberg_table")
  
  timeTravelDF.show()
  println(s"Data retrieved for snapshot: $snapshotId")
}

We can find the snapshot ID using:

val historyDF = spark.sql("SELECT * FROM spark_catalog.default.iceberg_table.history")
historyDF.show()

Once we have the snapshot ID, pass it to timeTravelBySnapshot() to query the data as of that snapshot.

Example: querying an Iceberg table as of a timestamp

def timeTravelByTimestamp(spark: SparkSession, timestamp: String): Unit = {
  val timeTravelDF = spark.read
    .option("as-of-timestamp", timestamp)
    .table("spark_catalog.default.iceberg_table")
  
  timeTravelDF.show()
  println(s"Data retrieved as of timestamp: $timestamp")
}
// Example timestamp format: "2024-10-16T15:00:00.000Z"
// To execute:
timeTravelByTimestamp(spark, "2024-10-16T15:00:00.000Z")

In this way we can to handle time travel queries using Apache Iceberg and Spark.