Apache Iceberg – Codeswarm

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.

Introduction

Partitioning helps improve query performance by reducing the amount of data scanned during read operations. Iceberg allows dynamic partitioning and supports various partition types such as identity, bucket, and truncate.

Example: partitioning data by a column (e.g., Age)

Let’s partition the table by a column like age (bucket partitioning).

def partitionTable(spark: SparkSession): Unit = { // Create a new partitioned table spark.sql( """ |CREATE TABLE IF NOT EXISTS spark_catalog.default.iceberg_partitioned_table ( | id BIGINT, | name STRING, | age INT |) USING iceberg | PARTITIONED BY (bucket(5, age)) |""".stripMargin) println("Partitioned Iceberg table created!") } // To execute: partitionTable(spark)

Here, we’ve partitioned the table using the age column into 5 buckets. This is useful for optimizing queries when filtering based on age.

Query example with partitioned table:

When you query partitioned data, Iceberg only scans the partitions relevant to the query:

def queryPartitionedTable(spark: SparkSession): Unit = { val partitionedDF = spark.sql("SELECT * FROM spark_catalog.default.iceberg_partitioned_table WHERE age > 25") partitionedDF.show() } // To execute: queryPartitionedTable(spark)

Tag: Apache Iceberg

Apache Iceberg and time travel queries with Scala/Spark

Apache Iceberg partitioning data with Scala/Spark

Introduction