Iceberg Tables POC

Introduction

Apache Iceberg (Apache Iceberg - Apache Iceberg™ ) is an open format for (big) analytic datasets. It is especially optimized for cloud storage (e.g. S3) and provides a familiar SQL table interface supporting ACID transactions.

In particular it supports update, delete and upsert operations, all while integrating with compute engines such as presto, spark etc. Other features include hidden partitioning and time-travel queries.

While there are other technologies that have similar features, iceberg is better integrated with the AWS ecosystem and in particular both in Glue (Using the Iceberg framework in AWS Glue - AWS Glue ) and Athena (Query Apache Iceberg tables - Amazon Athena ) supporting not only reading but also writing data.

Within the Synapse data warehouse, we make use of append only tables defined in the glue catalog and stored in S3 using the parquet format for efficient querying by Athena and other query engines. This works well with log like operations (e.g. events such as downloads) and we use the same approach for tracking changes in objects (e.g. snapshots of entities).

The purpose of this POC is to experiment with iceberg tables and see how it works.

We have a nodesnapshots table in glue that stores a copy of entities taken at various points in time and includes additional de-normalized information for efficient querying (e.g. is the entity public, the project it belongs to etc).

For the purpose of this POC, I pushed a dedicated set of events that emulates the evolution of changes of two nodes within 3 separate days, the data in the nodesnapshots looks something like the following (I omitted some of the columns for brevity):

We can see that nodeOne was created on 2023-04-08, updated on 2023-04-09 and deleted on 2023-04-10. NodeTwo was created on 2023-04-08, updated on 2023-04-09 and updated again on 2023-04-10. Basically the current state of the data should only include nodeTwo.

Creating the initial table

We can use apache iceberg tables directly from Athena starting from defining the table:

This creates an empty table at the given S3 location. Now for the purpose of the POC we want to start with the initial state of the DB and then emulate potential incremental updates. We can run a query to insert into the table the first snapshot for all the nodes:

Note that we use a window function to partition over each node and take only the first row according to the timestamp of the snapshot (e.g. the first snapshot). We use that data to “MERGE” into the iceberg table, since the table is empty the merge is not necessary but we show how an upsert could be done. In this case the only case that “matches” is the last condition (WHEN NOT MATCHED) since the table is empty. After executing the update we can query the table:

Updating the table

Now we can emulate an update to the table data taking a specific timeframe (e.g. just the day following the create) and considering only the last snapshot for the day:

Note that we now integrated a delete operation, if the id exists and the change type in the source was a “DELETE” we delete the record in the target. On day 2023-04-09 both nodes were updated, so we can query the table again:

We can see that it (as expected) now reflects the data as seen in the node snapshot table after the first update to both nodes (note the is_public flip and modified_on).

Finally, we update the table after “processing” the 3rd day (same query as before but with the snapshot_date changed):

If we now query the table we only have one record (on the 10th the first node was deleted and the second node was updated again):

Time-Travel Queries

Now we can even query the metadata of the table itself, for example we can see the transactions on the table:

And we can perform time-travel queries using the special syntax FOR TIMESTAMP AS OF TIMESTAMP: