Managing parallel development in extensive dbt projects is sometimes difficult. The complexity quickly escalates when multiple developers introduce simultaneous changes, especially when it which affects each others model.

The crucial question is: How can we reliably test individual changes without conflicts or the danger of compromising production data?

The effective approach I investigated involves SQLMesh, a data transformation tool that offers a powerful, native feature: Virtual Data Environments.

What are Virtual Data Environments?

The foundational principle is a sophisticated, yet simple, separation of concerns, brilliantly implemented using database views:

1. The Physical Layer: This layer holds the materialized objects, the actual logic.
2. The Virtual Layer: This layer is made up exclusively of database views positioned atop the physical layer objects.

To know more on Sqlmesh virtual environments please read the below article, its very detail

Virtual Data Environments

This design is transformative, enabling every developer to branch off the main production environment and instantly spin up a dedicated, isolated development/testing environment. Importantly, this happens without hindering other developers work or affecting live production data.

Testing SQLMesh with dbt: A POC

Given that SQLMesh is fully capable of running dbt projects, I was able to smoothly incorporate these virtual environments into a standard dbt workflow.

For this POC, I utilised the standard jaffle-sl-template project, which includes:

• Sample e-commerce data seeding.
• Raw data staging.
• Building core customer and order data marts.

Initial Project Setup Steps

Clone the repository and ensure our Snowflake connection credentials set correctly.

set up your Python environment:

python -m venv .venv
source .venv/bin/activate
pip install dbt-snowflake

load the raw data tables:

dbt seed

Finally, materialize the initial staging and mart models:

dbt run

Implementing Development Isolation with SQLMesh

With the dbt project operational, the goal is to utilize SQLMesh to establish isolated virtual environments for two distinct development tasks.

1. Installing SQLMesh and nesessary libraries

pip install "sqlmesh[snowflake]"
pip install "sqlmesh[dbt]"

2. Initializing the dbt Project for SQLMesh

Execute the sqlmesh init command from the dbt project root, selecting the dbt template:

sqlmesh init -t dbt

This command generates the sqlmesh.yaml configuration file. A key configuration to be observed in this file is:

virtual_environment_mode: dev_only

This setting ensures virtual environments are enabled only for development work, preserving the production environment’s integrity.So the production code still continues to work in dbt workflow. The other option is ‘full’ .

3. Creating the First Environment: feature-1234

Say a developer starts a new feature, branches their Git repository, and establishes their isolated environment using sqlmesh’s command

sqlmesh plan <environment name>

sqlmesh plan feature-1234

Upon successful execution, new schemas for phsysical and virtual layers will be visible in the database.

The Architectural Breakdown (for feature-1234):

• Physical Layer: SQLMesh creates physical tables, typically prefixed with sqlmesh__, which contain the actual data.
• Virtual Layer: This is the view layer that points to it respective physical layer object . New schemas will appear, such as:
• LIVE_STAGING__FEATURE_1234
• LIVE_MARTS__FEATURE_1234

Importantly, every object within this virtual layer is a database view. Inspecting the view’s definition will show it is simply a SELECT * from the corresponding physical object.

4. Creating the Second Environment: feature-5678

Now say a second developer now begins work on a completely separate feature:

sqlmesh plan feature-5678

This command creates another set of schemas for feature-5678,.

The feature-5678 environment introduces only new virtual schemas whose views still point to shared physical tables until code changes require new snapshots.

If one examines a view in this new environment (e.g., in LIVE_STAGING__FEATURE_5678), its code will reference the exact same physical object that the feature-1234 environment references. This shared reference to unchanged data is the basis for the efficiency of virtual environments.

The Real Test: Isolated Model Changes

Let’s check how this mechanism guarantees protection for parallel work streams.

Developer A (in feature-1234) needs to modify stg_products.sql to add a new column: product_category (classifying products as ‘menu_item’ or ‘other’).

The developer updates the model code and re-runs the plan command:

sqlmesh plan feature-1234

The Process in Action:

When SQLMesh detects the code change, it generates a new physical model for stg_products with a unique fingerprint hash (e.g., a table ending with 374258167. The one ending with 417388839 is the old model without the new column.).

This new physical table incorporates the updated logic and the added column.

• The stg_products view in the feature-1234 schema is automatically updated to point to this newly created, modified physical model.

• Crucially, the stg_products view in the feature-5678 schema continues to point to the original physical model, remaining completely unaffected by the change in feature-1234.

Thus feature-1234 and feature-5678 can be tested independently without affecting each other.

Conclusion:

For scenarios involving multiple concurrent contributors, SQLMesh’s virtual environment feature can be an really helpful asset in effectively eliminating the risk of accidental dependencies and data corruption.

Considering the recent merger between Fivetran and dbt Labs, and Fivetran’s prior acquisition of SQLMesh, it’s an interesting possibility that this isolated, virtual data environments feature soon become a native or more tightly integrated feature within the dbt ecosystem itself.

Disclaimer: I used AI tools like ChatGPT just to help with grammar and phrasing. The content, ideas, and all the work in this article are 100% my own.

Unlock faster analytics by mastering micro-partition pruning, clustering strategies, and search optimisation in Snowflake.

Understanding micro-partitions and pruning

Snowflake automatically divides all table data into micro-partitions. Micro-partitions are contiguous storage units that typically hold between 50 MB and 500 MB of uncompressed data (though Snowflake always stores this data in a compressed format). Each micro-partition organises groups of rows in a columnar structure, enabling highly granular pruning.

What makes this powerful is the metadata Snowflake stores for every micro-partition. This includes the range of values for each column, the number of distinct values, and other properties that help optimize and speed up query processing.

Let’s Look at an example Table

Below is the conceptual representation of how the above data might look when stored in micro-partitions in snowflake.

This data is stored across three micro-partitions, with the rows divided equally between each. The first three rows are stored in the first micro-partition, the following three rows in the second micro-partition, and the last three in the third micro-partition, as shown above.

Pruning in Snowflake

Because Snowflake tracks metadata for each micro-partition (like the range of values for every column), it can optimize queries by scanning only the micro-partitions that might contain the data we need. This process of excluding unnecessary partitions is called pruning.

For example, if we query for ID = 100 in the above table,Snowflake only reads micro-partitions 1 and 2, skipping the third entirely.

But if we filter by Date = 'Jan-10', since that date appears in all micro-partitions, the query must scan all of them. However, if micro-partitions were sorted by date instead of ID, the query could potentially scan fewer partitions, making it more efficient.

Seeing Pruning in Action

Let’s try this with a large dataset from Snowflake’s sample data:

CREATE TABLE LINEITEM AS 
SELECT * FROM SNOWFLAKE_SAMPLE_DATA.TPCH_SF1000.LINEITEM;

This table has over 6 billion rows. Running a query like below can take a long time if the table isn’t well-clustered, because Snowflake may need to scan all partitions.

SELECT * FROM LINEITEM WHERE L_SHIPDATE = '1992-10-30';

The above query ran for 1minute and 23 seconds in a X-Small warehouse.

If we click on the query id and see the statistics , we see the query has to scan all the 9,231 partitions.

Optimising with Clustering

Run the below query to check how well the table is clustered based on L_SHIPDATE.

SELECT SYSTEM$CLUSTERING_INFORMATION('LINEITEM', '(L_SHIPDATE)');

{
  "cluster_by_keys" : "LINEAR(L_SHIPDATE)",
  "total_partition_count" : 9234,
  "total_constant_partition_count" : 0,
  "average_overlaps" : 9233.0,
  "average_depth" : 9234.0,
  "partition_depth_histogram" : {
    "00000" : 0,
    "00001" : 0,
    "00002" : 0,
    "00003" : 0,
    "00004" : 0,
    "00005" : 0,
    "00006" : 0,
    "00007" : 0,
    "00008" : 0,
    "00009" : 0,
    "00010" : 0,
    "00011" : 0,
    "00012" : 0,
    "00013" : 0,
    "00014" : 0,
    "00015" : 0,
    "00016" : 0,
    "16384" : 9234
  },
  "clustering_errors" : [ ]
}

The total_constant_partition_count is 0, meaning the table is not well clustered (a larger value means the clustering is better).

The average_overlaps value is almost the same as the total_partition_count value, which means that nearly all micro-partitions overlap, and consequently, the table is not well clustered (a lower value means that clustering is better).

The average_depth value is the same as the total_partition_count value, meaning the table is not well clustered. A smaller value means the clustering is better.

Adding a clustering key can significantly improve query performance.

Once we define a clustering key, Snowflake will periodically reorganise data in the background to optimize micro-partitioning.

Lets cluster the LINEITEM table by by L_SHIPDATE:

ALTER TABLE LINEITEM CLUSTER BY (L_SHIPDATE);

The clustering should happen in the backend. We can monitor the status by running the below table function.

select *
  from table(information_schema.automatic_clustering_history(
  date_range_start=>dateadd(D, -1, current_date),
  table_name=>'TPCH_COPY.PRUNING.LINEITEM'));

We should see something like below once the clustering is completed.

Lets now run the same query again.

select * from LINEITEM where L_SHIPDATE ='1992-10-30';

now the completed in just 4.2 seconds. The query benefits from partition pruning — out of 9,327 partitions, only 10 are scanned.

When to Use Clustering

Keep these criteria in mind when deciding on clustering:

• The table contains a large number of micro-partitions, so pruning will be significant.
• Queries that select a small portion of rows, retrieving a small number of micro-partitions.
• If most of the queries filter by the same column (often a date or time period) ,adding that column to the cluster key can greatly benefit from partition pruning.
• The number of distinct values in the clustering column is large enough for pruning, but small enough for Snowflake to colocate rows with the same key in the same micro-partition.

Adding clustering keys and allowing Snowflake to maintain them automatically in the background incurs cost. So, understand the types of queries most often executed on the table.

Search Optimisation in Snowflake

Let’s say we want to filter the LINEITEM table based on the L_ORDERKEY column.

To improve the performance of queries that filter on L_ORDERKEY, we can’t cluster the LINEITEM table again, since it’s already clustered by L_SHIPDATE. Instead, we can take advantage of Snowflake’s search optimisation.

The search optimisation service boosts query performance when filtering data. When search optimisation is added to a table, Snowflake creates a search access path in the background. This access path keeps track of which values of the table’s columns reside in each micro-partition. As a result, micro-partition pruning becomes much more effective when querying the table with filters. Snowflake also maintains the search access path automatically as data is inserted, updated, or deleted.

We can add search optimisation to a table without specifying columns, in which case Snowflake builds the access path for all eligible columns. However, this can be costly. It’s usually better to add search optimisation only to the columns that are frequently used in filters.

lets run the below query to add search optimisation to the L_ORDERKEY column of the LINEITEM table

ALTER TABLE LINEITEM ADD SEARCH OPTIMIZATION ON EQUALITY(L_ORDERKEY);

Run the below query to monitor the progress

SHOW TABLES LIKE 'LINEITEM';

Now, let’s look at the difference in query performance before and after enabling search optimisation

SELECT L_ORDERKEY FROM LINEITEM WHERE L_ORDERKEY = 5461081319;

The query ran for 31 secs before search optimisation. It has to scan almost all micro-partitions.

Same query ran for only 991 ms after search optimisation

As we can see in the statistics pane, the number of scanned partitions is 3, while the total number of partitions is 9399. This indicates that search optimisation improved query performance by using micro-partition pruning.

Tips for adding search optimization:

• The table has a large number of micro-partitions.
• Queries use equality filters.
• The filter in the query returns only a small subset of rows.
• The table is already clustered, but the columns used in filters are not the clustering key.

Conclusion

By understanding how micro-partitions work, applying clustering where it helps, and using search optimisation for specific needs, we can make your queries run much faster. Just remember, both clustering and search optimisation can add extra costs, so use them where they make the most impact.

References

• Snowflake Data Engineering (O’Reilly) — Chapter 8: Clustering and Pruning
• Snowflake Documentation: Clustering & Micro-Partitions
• Snowflake Documentation: Clustering Keys
• Snowflake Documentation: Search Optimization Service

Disclaimer: I used AI tools like ChatGPT just to help with grammar and phrasing. The content, ideas, and all the work in this article are 100% my own.

In recent years, the lakehouse architecture has gained significant attention in the data engineering world. A lakehouse combines the best features of data lakes and data warehouses. It brings the ACID principles of data warehousing while supporting unstructured and semi-structured data, all stored in cost-effective storage solutions like AWS S3.

The technology around lakehouse is evolving rapidly, with numerous tools emerging to query data from lakehouses efficiently, reducing latency. At its core, the lakehouse architecture is known for its decoupled nature of storage and compute. This means you can use inexpensive storage solutions and pair them with any query engine to process the data. This flexibility allows organizations to switch query engines in the future based on their needs, avoiding vendor lock-in.

The Promise of Decoupling: Are We There Yet?

One of the most significant benefits of the lakehouse architecture is the ability to switch query engines without being locked into a single tool. However, are we truly realising this benefit? The answer is: not entirely.

Modern data engineering is not just about running ad-hoc queries. It involves building and maintaining pipelines that run on a schedule, whether in batch or streaming mode. These pipelines could be ETL (Extract, Transform, Load) or ELT (Extract, Load, Transform).

For example, in an ELT pipeline, raw data is first loaded into an open table format like Apache Iceberg using engine like apache spark. From there, layers of transformations are applied to prepare the data for use in BI tools. Typically, these transformations are written as SQL queries.

Consider the following example of a transformation pipeline:

Bronze to Silver Layer (Spark SQL):

INSERT INTO silver.cleaned_transactions
SELECT 
    transaction_id,
    order_date,
    customer_id,
    product_id,
    quantity,
    unit_price,
    quantity * unit_price as total_amount,
    UPPER(payment_method) as payment_method,
    source_system,
    CASE 
        WHEN quantity > 0 AND unit_price > 0 THEN TRUE 
        ELSE FALSE 
    END as is_valid
FROM bronze.raw_transactions;

Silver to Gold Layer (Spark SQL):

INSERT INTO gold.daily_metrics
SELECT 
    order_date,
    COUNT(DISTINCT transaction_id) as total_orders,
    COUNT(DISTINCT customer_id) as total_customers,
    SUM(total_amount) as total_revenue,
    AVG(total_amount) as avg_order_value
FROM silver.cleaned_transactions
WHERE is_valid = TRUE
AND date_add(order_date, 7) BETWEEN @start_date AND @end_date
GROUP BY order_date;

In practice, tools like DBT(Data Build Tool) are often used to manage these SQL transformations and streamline workflows. However, with time hundreds of similar pipelines are deployed in production using a specific query engine (e.g., Spark) dialect, a challenge arises: what happens if the organisation decides to switch to a different query engine, such as DuckDB or Snowflake or Trino?

This would require rewriting all the SQL queries to match the syntax of the new engine. The effort involved in such a migration can be enormous, negating the flexibility promised by the lakehouse architecture.Can we decouple the sql dialect from query engines?

SQLMesh: A Solution for True Decoupling

SQLMesh is an open-source transformation tool that we can use to address this problem. Similar to DBT but with added intelligence, SQLMesh leverages the sqlglot library, which can parse, interpret, and transpile SQL code across different dialects. This makes it possible to decouple your pipelines from a specific query engine, allowing seamless transitions between engines without breaking anything.

SQLMesh is not just a transformation tool — it’s a robust DataOps platform built to handle the complexities of modern data engineering. While DBT has been a pioneer in the transformation space, it struggles to scale with increasing data and organizational complexity. SQLMesh, on the other hand, was designed from the ground up to address these challenges.

Note : Sqlmesh has lots of cool features but in this article we are mainly focusing on the dialect transpilation . Please refer documentation for more details.

Using SQLMesh to Decouple Compute

To use SQLMesh, we define your transformations in its specific format. Let’s revisit the earlier example and convert it into SQLMesh models.

Bronze Layer: bronze.raw_transactions

This is a seed model to load raw data from a CSV file.

MODEL (
  name bronze.raw_transactions,
  kind SEED (
    path '../seeds/seed_data.csv'
  ),
  columns (
    transaction_id VARCHAR,
    order_date DATE,
    customer_id VARCHAR,
    product_id VARCHAR,
    quantity INTEGER,
    unit_price FLOAT,
    payment_method VARCHAR,
    source_system VARCHAR
  ),
  grain (transaction_id)
);

Silver Layer: silver.cleaned_transactions

MODEL (
  name silver.cleaned_transactions,
  kind INCREMENTAL_BY_TIME_RANGE (
    time_column order_date
  ),
  start '2024-01-01',
  cron '@daily',
  grain (transaction_id)
);

SELECT 
    transaction_id,
    order_date,
    customer_id,
    product_id,
    quantity,
    unit_price,
    quantity * unit_price as total_amount,
    UPPER(payment_method) as payment_method,
    source_system,
    CASE 
        WHEN quantity > 0 AND unit_price > 0 THEN TRUE 
        ELSE FALSE 
    END as is_valid
FROM bronze.raw_transactions
WHERE order_date BETWEEN @start_date AND @end_date;

Gold Layer: gold.daily_sales_summary

MODEL (
  name gold.daily_sales_summary,
  kind INCREMENTAL_BY_TIME_RANGE (
    time_column order_date
  ),
  start '2024-01-01',
  cron '@daily',
  grain (transaction_id)
);

SELECT 
    order_date,
    COUNT(DISTINCT transaction_id) as total_orders,
    COUNT(DISTINCT customer_id) as total_customers,
    SUM(total_amount) as total_revenue,
    AVG(total_amount) as avg_order_value
FROM silver.cleaned_transactions
WHERE is_valid = TRUE
AND date_add(order_date, 7) BETWEEN @start_date AND @end_date
GROUP BY order_date;

Transpiling SQL Across Engines

Below is the SQLmesh configuration file, where we define the connection details along with other details:

gateways:
  local_spark_connection:
    connection:
      type: spark
    state_connection:
      type: duckdb
      database: spark_state.db
  my_snowflake_connection:
    connection:
      type: snowflake
      account: idapznw-wq29399
      user: DEMO_USER
      password: << password here >>
      database: DEMO_DB
      warehouse: COMPUTE_WH
    state_connection:
      type: duckdb
      database: snowflake_state.db
  duckdb_connection:
    connection:
      type: duckdb
      catalogs:
        persistent: 'local.duckdb'
        ephemeral: ':memory:'
    state_connection:
      type: duckdb
      database: duckdb_state.db

default_gateway: local_spark_connection

For example, in the provided configuration, three connections are defined:

1. local_spark_connection: A connection to a Spark environment.
2. my_snowflake_connection: A connection to a Snowflake environment.
3. duckdb_connection: A connection to a DuckDB environment.

The default_gateway in the configuration is set to local_spark_connection, meaning that by default, SQLMesh will deploy the models to the Spark environment. When you run the sqlmesh plan command, SQLMesh processes the models and generates SQL queries tailored to the default gateway's dialect (in this case, Spark SQL). For example, if the gold.daily_sales_summary model is deployed to Spark, the query might look like this:

SELECT
  `cleaned_transactions`.`order_date` AS `order_date`,
  COUNT(DISTINCT `cleaned_transactions`.`transaction_id`) AS `total_orders`,
  COUNT(DISTINCT `cleaned_transactions`.`customer_id`) AS `total_customers`,
  SUM(`cleaned_transactions`.`total_amount`) AS `total_revenue`,
  AVG(`cleaned_transactions`.`total_amount`) AS `avg_order_value`
FROM `catalog`.`silver`.`cleaned_transactions` AS `cleaned_transactions`
WHERE
  `cleaned_transactions`.`is_valid` = TRUE
  AND DATE_ADD(`cleaned_transactions`.`order_date`, 7) <= CAST('1970-01-01' AS DATE)
  AND DATE_ADD(`cleaned_transactions`.`order_date`, 7) >= CAST('1970-01-01' AS DATE)
GROUP BY
  `cleaned_transactions`.`order_date`;

Here, the query uses Spark-specific syntax, such as DATE_ADD.

Now, if you decide to switch the default_gateway to my_snowflake_connection in the yaml file and run the sqlmesh plan command again, SQLMesh recognizes that the target environment is Snowflake. It automatically transpiles the Spark SQL query into Snowflake SQL, ensuring compatibility with Snowflake's dialect. For example, the same query for the gold.daily_sales_summary model would be rendered as:

SELECT
  "cleaned_transactions"."order_date" AS "order_date",
  COUNT(DISTINCT "cleaned_transactions"."transaction_id") AS "total_orders",
  COUNT(DISTINCT "cleaned_transactions"."customer_id") AS "total_customers",
  SUM("cleaned_transactions"."total_amount") AS "total_revenue",
  AVG("cleaned_transactions"."total_amount") AS "avg_order_value"
FROM "catalog"."silver"."cleaned_transactions" AS "cleaned_transactions"
WHERE
  "cleaned_transactions"."is_valid" = TRUE
  AND DATEADD(DAY, 7, CAST(CAST("cleaned_transactions"."order_date" AS TIMESTAMP) AS DATE)) <= TRY_CAST('1970-01-01' AS DATE)
  AND DATEADD(DAY, 7, CAST(CAST("cleaned_transactions"."order_date" AS TIMESTAMP) AS DATE)) >= TRY_CAST('1970-01-01' AS DATE)
GROUP BY
  "cleaned_transactions"."order_date";

Notice how the DATE_ADD function from Spark is automatically converted to Snowflake's DATEADD function, which uses a different syntax. This transpilation happens seamlessly, saving you from manually rewriting the query.

Similarly, if you switch the default_gateway to duckdb_connection and run the sqlmesh plan command, SQLMesh will transpile the query into DuckDB's dialect. For example:

SELECT
  "cleaned_transactions"."order_date" AS "order_date",
  COUNT(DISTINCT "cleaned_transactions"."transaction_id") AS "total_orders",
  COUNT(DISTINCT "cleaned_transactions"."customer_id") AS "total_customers",
  SUM("cleaned_transactions"."total_amount") AS "total_revenue",
  AVG("cleaned_transactions"."total_amount") AS "avg_order_value"
FROM "catalog"."silver"."cleaned_transactions" AS "cleaned_transactions"
WHERE
  "cleaned_transactions"."is_valid" = TRUE
  AND CAST("cleaned_transactions"."order_date" AS DATE) + INTERVAL 7 DAY <= TRY_CAST('1970-01-01' AS DATE)
  AND CAST("cleaned_transactions"."order_date" AS DATE) + INTERVAL 7 DAY >= TRY_CAST('1970-01-01' AS DATE)
GROUP BY
  "cleaned_transactions"."order_date";

In this case, the DATE_ADD function is replaced with + INTERVAL 7 DAY, which is the syntax used by DuckDB.

Why This Matters

This automatic transpilation ensures that your pipelines remain functional regardless of the query engine you use. You can switch between Spark, Snowflake, DuckDB, or any other supported engine without rewriting your SQL queries. SQLMesh handles the heavy lifting of adapting the SQL syntax to match the target engine’s requirements, allowing you to fully leverage the decoupling benefits of the lakehouse architecture.

By abstracting away the complexities of SQL dialect differences, SQLMesh makes it easier to build and maintain flexible, future-proof data pipelines. Whether you’re migrating to a new query engine or running a multi-engine setup, SQLMesh ensures that your transformations remain consistent and reliable.

Disclaimer

I used AI tools like ChatGPT just to help with grammar and phrasing. The content, ideas, and all the work in this article are 100% my own.

Introduction

Apache Iceberg tables include a layer of metadata that helps data engineers to optimize the table for better performance. This metadata is collected and stored whenever an operation occurs in the table, such as append or overwrite. As a result, accessing this metadata is less costly than scanning the entire table.

This metadata provides excellent insights into the data in the table, such as:

• The number and size of data files.
• Partitioning details of the table.
• Frequency of operations performed on the table.
• Evolution of table and partition columns over time.
• Snapshot history of the table, and much more.

However, understanding these metrics can be difficult at times. With so many tables and so much metadata information, it can be overwhelming to make sense of it all. In this blog, I will explain how I used Streamlit to create an insightful dashboard that visualizes these metrics and serves as a useful tool. This dashboard can be used in production to monitor and maintain Apache Iceberg tables. It also helps to proactively optimize tables for faster performance and resolve existing issues.

The Streamlit dashboard I created ,the main page of the dashboard provides an overall summary of metrics for all tables.

Total Counts: Displays total counts of tables, active files, and cumulative sizes.

Real-Time Monitoring: Includes a last updated timestamp.

Left side of the page lists of all tables in the namespace .Users can search or click on each table to view its metrics individually.

Table Summary Metrics

The first section of the dashboard provides a summary of the table’s key metrics:

Active/Total Files & Active/Total Storage

• These cells provide insight into the number of active and inactive files in the table.
• Files are marked Inactive whenever the table is updated, deleted, or undergoes a similar operation. Since Iceberg data lakes are immutable, operations like updates or deletes don’t modify the existing files. Instead, they create new files with the updated records and update the metadata to reflect this change.
• Inactive files are kept rather than deleted immediately to support time travel functionality. However, maintaining a balance in managing inactive files is essential.

Total Records

• Displays the total number of records in the table.

First Snapshot

• Shows the first accessible snapshot for the table.

Last Snapshot

• Indicates the last modification date of the table.

Total Snapshots

• Shows the total number of snapshots available for the table. Running a vacuum job to remove inactive files may reduce this count.

Average File Size

• Displays the average size of data files in the table.

Daily Growth Metrics

The next section provides daily growth metrics to track how the table has grown over time.Beyond static metrics, the daily growth section gives a detailed view of how our Iceberg tables evolve over time.

Key Features:

Daily Growth Statistics:

Tracks the daily growth of records, files, and table size.

Timeframe Selector:

• Users can analyze data over specific timeframes such as 1 day, 5 days, 1 month, 6 months, 1 year, or all time.
• Highlights the difference from the previous day using green (increase) or red (decrease).

Growth Visualization Charts:

• Records Growth (Line Chart): Displays the total records and new records added each day. The solid line represents cumulative totals, while the dashed line shows daily additions.
• Files Growth (Line Chart): Displays the total number of files and files added daily, helping to identify file creation patterns.
• Size Growth (Line Chart): Tracks the total table size and daily size increases. This is crucial for understanding storage trends and optimising efficiency.

Purpose and Benefits:

• Daily Monitoring: Enables teams to monitor the daily growth of records, files, and table size, offering a detailed view of the table’s evolution.
• Trend Analysis: Helps users easily identify growth patterns or anomalies, such as record count spikes or unusual file additions.

Snapshot Analysis

This section focuses on snapshot history analysis, providing detailed insights into changes and operations performed over time. It also allows users to evaluate snapshots before time travel to a specific snapshot.

Key Features:

Snapshot History Analysis:

• Start and End Date Selector: Filters snapshot activities within a specific time range.
• Filter by Operation Type: Users can filter data by operations such as append and overwrite.
• Records Added and Deleted (Line Chart): Shows the number of records added and deleted for each snapshot. Added and deleted records are plotted separately for a clear view of data growth or reduction patterns.
• Operation Timeline (Scatter Plot): Visualizes operations (e.g., append, overwrite, delete) with distinct markers for each type, making it easy to analyse their order and frequency.

Choose Snapshot for Time Travel:

• Snapshot Dropdown: Allows users to select a specific snapshot.
• Operational Details: Displays the operation type and the number of records added or deleted for the selected snapshot.
• Time Travel Button: Enables users to revert to a specific snapshot directly from the dashboard.

Operation Counts Over Time (Bar Chart):

• Displays the daily count of each operation type, helping users understand the frequency and distributionof operations over a given period.

Partition Analysis

The Partition Analysis section provides insights into how data is distributed across partitions in an Iceberg table.

Key Features:

Partition Scheme Summary:

• Offers an overview of the partitioning scheme and its evolution over time.
• Displays the number of records for each partition scheme.

Partition Size and Record Count (Bar and Line Chart):

• Bar Chart (Partition Size): Blue bars show the total size of each partition.
• Record Count (Line Chart): A red line overlaid on the bar chart represents the record count for each partition.
• Helps identify skewed partitions for optimization.

File Count Distribution (Pie Chart):

Displays the distribution of files by partition, with each segment representing a specific partition

File Size Analysis

Managing file sizes is critical to maintaining query performance and minimising I/O overhead. This section focuses on analysing file sizes in detail.

Key Features:

Overall File Size Deviation (Left Scatter Plot):

• Each dot represents an individual file.
• A red dashed line shows the average file size, helping to identify outliers that are too small or too large.

Partition-Specific File Size Distribution (Right Scatter Plot):

• Users can select a specific partition from a dropdown to drill down into its file size distribution.
• Similar to the overall scatter plot, this chart shows file sizes within the selected partition.

File Compaction Job Trigger:

• Run Compaction Job Button: Allows users to consolidate small files into larger ones, improving query performance and storage efficiency.

This dashboard transforms Apache Iceberg’s rich metadata into a practical, easy-to-use tool for monitoring, optimising, maintaining and take action on tables. By visualising key metrics, it simplifies complex operations and empowers data engineers to take timely action for better performance and efficiency.There are still more metrics that can be derived from the metadata, which can be transformed into useful visualisations to provide even deeper insights into an iceberg table.

We’ll see the code to get the metrics and this dashboard in the next blog.

References

• Apache Iceberg: The Definitive Guide
• Monitoring Apache Iceberg with AWS Lambda, Glue, and CloudWatch

Disclaimer

I used AI tools like ChatGPT just to help with grammar and phrasing. The content, ideas, and all the work in this article are 100% my own.

Table of Contents

1. Introduction
2. Dataset Overview
3. Comparison Results
4. Key Observations
5. Data and Setup Process
6. Break Down of Queries
7. Final Verdict

Introduction

When working with large datasets, query performance is critical for optimizing time and resources. In this article, we compare the performance differences between querying a native Snowflake table and querying an Iceberg table, both using Snowflake.

Snowflake’s proprietary storage format is designed for fast querying and efficient data management, while Iceberg offers flexibility with open-source table formats, often stored on external systems like S3. By running similar queries on both types of tables from snowflake, we’ll highlight the differences in:

• Query Execution Time: How quickly queries return results.
• Data Scanned: The amount of data read from storage.
• Partitioning Efficiency: How efficiently data is retrieved.

Note: To ensure accurate testing, I turned off the cache in Snowflake by running:

ALTER SESSION SET USE_CACHED_RESULT = FALSE;

Dataset Overview

For this performance comparison, I used a dataset containing 1 billion rows of temperature measurements across various weather stations. You can access the dataset from the 1 Billion Row Challenge GitHub repository.

Each row has a station_name and temperature, formatted as <station_name>;<temperature>, with the temperature having one decimal place.

Hamburg;12.0
Bulawayo;8.9
Palembang;38.8
St. John's;15.2

The dataset is around 15 GB, but after loading it into Snowflake and Iceberg, the sizes were compressed to 2.6 GB for the native Snowflake table and 3.3 GB for the Iceberg table on S3.

Before we unravel the implementation behind this comparison, let’s jump right into the results and see how Snowflake native tables stack up against Iceberg tables queried in Snowflake.

Performance Overview: Snowflake vs. Iceberg

Key Observations

Query Execution Time: Across a variety of queries — including simple SELECTs, aggregations, and range filters — Snowflake’s native tables consistently delivered faster performance, often executing 2–5 times faster than Iceberg tables. For instance, the Simple SELECT query returned results in just 0.483 seconds on Snowflake, compared to 0.622 seconds on Iceberg. This performance gap widened further with more complex queries like aggregations and range filters, where Snowflake’s optimized architecture clearly demonstrated its efficiency.

Let’s Now Dig Deeper: How We Performed This Comparison

Step 1: Loading Data into Snowflake Native Storage

To load the dataset into Snowflake’s native storage, we begin by logging in to Snowflake using SnowSQL and then uploading the file to an internal stage before loading it into the table.

1. Login to Snowflake Account from Local Machine Using SnowSQL: Open your terminal and log in to your Snowflake account using SnowSQL.

snowsql -a <account> -u <username> -r <role> -w <warehouse>

2. Upload the File to Snowflake’s Internal Stage: Use the PUT command to upload the dataset file to the internal stage for loading.

PUT file:///1brc/data/measurements.txt @my_int_stage AUTO_COMPRESS=TRUE;

3.Create the Table and Load Data: Once the file is uploaded, create the necessary table and load the data from the stage into the table.

CREATE OR REPLACE TABLE measurements (
  location VARCHAR(50),
  temperature NUMBER
);

CREATE OR REPLACE FILE FORMAT measurement_format
  TYPE = 'CSV'
  FIELD_DELIMITER = ';';

COPY INTO measurements
FROM @my_int_stage
FILE_FORMAT = (FORMAT_NAME = measurement_format)
ON_ERROR = 'skip_file';

The structure of the table is shown below

Step 2: Creating and Querying an Iceberg Table in S3

Next, I created an Iceberg table for the same dataset, but this time stored externally on S3. I used PySpark to write the data into Iceberg format and linked it to Snowflake via the Polaris catalog.

1. Upload Data to S3 and Write as Iceberg Table:

import  pyspark.sql.functions as f
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType,StructField,StringType,IntegerType,DecimalType
import os

spark = SparkSession.builder.appName('iceberg') \
.config('spark.jars.packages', 'org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.4.1,software.amazon.awssdk:bundle:2.20.160,software.amazon.awssdk:url-connection-client:2.20.160') \
.config('spark.sql.extensions', 'org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions') \
.config('spark.sql.defaultCatalog', 'polaris') \
.config('spark.sql.catalog.polaris', 'org.apache.iceberg.spark.SparkCatalog') \
.config('spark.sql.catalog.polaris.type', 'rest') \
.config('spark.sql.catalog.polaris.header.X-Iceberg-Access-Delegation','vended-credentials') \
.config('spark.sql.catalog.polaris.uri','https://xxxx.ap-southeast-1.snowflakecomputing.com/polaris/api/catalog') \
.config('spark.sql.catalog.polaris.credential','xxxxx/xx=:xxxx+xxx=') \
.config('spark.sql.catalog.polaris.warehouse','my_polaris_catalog') \
.config('spark.sql.catalog.polaris.scope','PRINCIPAL_ROLE:data_engineer_role')\
.config('spark.sql.catalog.polaris.client.region','ap-southeast-1')\
.getOrCreate()

df = spark.read.text('/1brc/data/measurements.txt')

df = df.select(f.split(f.col("value"),";").alias('c1'))

df = df.select(f.col('c1')[0].alias('location'),f.col('c1')[1].cast(DecimalType(38,0)).alias('temperature'))

database_name = "onebrc"
table_name = "tbl_measurements"

df.write.format("iceberg") \
    .mode("overwrite") \
    .saveAsTable(f"{database_name}.{table_name}")

2. Integrate with Snowflake Polaris and create an iceberg table in Snowflake:

CREATE OR REPLACE EXTERNAL VOLUME iceberg_external_volume
   STORAGE_LOCATIONS = (
         (
            NAME = 'my-s3-ap-southeast-1',
            STORAGE_PROVIDER = 'S3',
            STORAGE_BASE_URL = 's3://lakehouse-8765374/',
            STORAGE_AWS_ROLE_ARN = 'arn:aws:iam::xxxxxx:role/sf_external_volume_role',
            STORAGE_AWS_EXTERNAL_ID = 'iceberg_table_external_id'
         )
      );


CREATE OR REPLACE CATALOG INTEGRATION my_polaris_int
  CATALOG_SOURCE = POLARIS
  TABLE_FORMAT = ICEBERG
  CATALOG_NAMESPACE = 'onebrc'
  REST_CONFIG = (
    CATALOG_URI = 'https://xxxx.ap-southeast-1.snowflakecomputing.com/polaris/api/catalog'
    WAREHOUSE = 'my_polaris_catalog'
  )
  REST_AUTHENTICATION = (
    TYPE = OAUTH
    OAUTH_CLIENT_ID = 'xxxxxx'
    OAUTH_CLIENT_SECRET = 'xxxxxxx'
    OAUTH_ALLOWED_SCOPES = ('PRINCIPAL_ROLE:data_engineer_role')
  )
  ENABLED = TRUE;


  CREATE or replace ICEBERG TABLE measurements_iceberg
  EXTERNAL_VOLUME='iceberg_external_volume',
  CATALOG='my_polaris_int',
  CATALOG_TABLE_NAME='tbl_measurements';

The structure of the table is shown below

Detailed Breakdown of Queries:

Below are the 10 queries that were run across both the Snowflake native table (measurements) and the Iceberg table(measurements_iceberg) to compare their performance in terms of runtime, data scanned, and partitions scanned.

1. Simple SELECT Query:

-- Snowflake Native Storage
SELECT * FROM measurements LIMIT 1000;

-- Iceberg Table in S3
SELECT * FROM measurements_iceberg LIMIT 1000;

2. Filter by Temperature Range:

-- Snowflake Native Storage
SELECT * 
FROM measurements
WHERE temperature >= 20.0 AND temperature <= 30.0;

-- Iceberg Table in S3
SELECT * 
FROM measurements_iceberg
WHERE temperature >= 20.0 AND temperature <= 30.0;

3. Aggregation by Location (AVG)

-- Snowflake Native Storage
SELECT location, AVG(temperature) AS avg_temperature 
FROM measurements 
GROUP BY location;

-- Iceberg Table in S3
SELECT location, AVG(temperature) AS avg_temperature 
FROM measurements_iceberg 
GROUP BY location;

4. Count Total Measurements

-- Snowflake Native Storage
SELECT COUNT(*) AS total_measurements 
FROM measurements;

-- Iceberg Table in S3
SELECT COUNT(*) AS total_measurements 
FROM measurements_iceberg;

5. Max Temperature by Location

-- Snowflake Native Storage
SELECT location, MAX(temperature) AS max_temperature 
FROM measurements 
GROUP BY location;

-- Iceberg Table in S3
SELECT location, MAX(temperature) AS max_temperature 
FROM measurements_iceberg 
GROUP BY location;

6. Min Temperature by Location

-- Snowflake Native Storage
SELECT location, MIN(temperature) AS min_temperature 
FROM measurements 
GROUP BY location;

-- Iceberg Table in S3
SELECT location, MIN(temperature) AS min_temperature 
FROM measurements_iceberg 
GROUP BY location;

7. Average Temperature Across All Locations

-- Snowflake Native Storage
SELECT AVG(temperature) AS avg_temperature 
FROM measurements;

-- Iceberg Table in S3
SELECT AVG(temperature) AS avg_temperature 
FROM measurements_iceberg;

8. Top 10 Hottest Locations (AVG Temperature)

-- Snowflake Native Storage
SELECT location, AVG(temperature) AS avg_temperature 
FROM measurements 
GROUP BY location 
ORDER BY avg_temperature DESC 
LIMIT 10;

-- Iceberg Table in S3
SELECT location, AVG(temperature) AS avg_temperature 
FROM measurements_iceberg 
GROUP BY location 
ORDER BY avg_temperature DESC 
LIMIT 10;

9. Temperature Variance by Location

-- Snowflake Native Storage
SELECT location, VAR_SAMP(temperature) AS temperature_variance 
FROM measurements 
GROUP BY location;

-- Iceberg Table in S3
SELECT location, VAR_SAMP(temperature) AS temperature_variance 
FROM measurements_iceberg 
GROUP BY location;

10. Min, Max, and Avg Temperature by Location

-- Snowflake Native Storage
SELECT 
    location,
    MIN(temperature) AS Min,
    MAX(temperature) AS Max,
    AVG(temperature) AS Avg
FROM measurements 
GROUP BY location 
ORDER BY location;

-- Iceberg Table in S3
SELECT 
    location,
    MIN(temperature) AS Min,
    MAX(temperature) AS Max,
    AVG(temperature) AS Avg
FROM measurements_iceberg 
GROUP BY location 
ORDER BY location;

My Final Verdict:

This comparison clearly shows that Snowflake’s native storage outperforms Iceberg in terms of query execution time and data retrieval efficiency. Snowflake’s native tables consistently returned results faster across a variety of query types, particularly for aggregation and filtering queries, where the difference was most pronounced.

However, Iceberg offers flexibility with its open-source format, allowing integration with external storage systems like S3. While Iceberg’s performance was generally slower, it may still be an attractive option for users who prioritize flexibility and open standards over raw query speed.

Moving forward, I plan to continue testing by applying several optimization techniques to Iceberg, including:

• Trying different compression settings to store the data optimally and see if query performance can be improved.
• Running OPTIMIZE queries to organize data more efficiently.
• Implementing Z-ordering to reduce the amount of data scanned and further optimize performance.

Additionally, I’ll also look into Snowflake-specific optimizations.

Ultimately, the choice between native Snowflake tables and Iceberg on S3 depends on your specific use case. For workloads requiring speed and efficiency, Snowflake’s native storage is the better choice. For workloads prioritizing flexibility and interoperability, Iceberg may be worth considering, especially with further optimizations.

Stay tuned for updates on these optimizations and further testing to close the performance gap.

Disclaimer: I used AI tools like ChatGPT just to help with grammar and phrasing. The content, ideas, and all the work in this article are 100% my own.

Lakehouse architecture is a modern approach that combines the best features of data warehouses and data lakes. It merges the flexibility and cost-effectiveness of a data lake with the reliability and performance of a data warehouse. In this article, I will guide you through implementing a lakehouse architecture using AWS S3, Glue Catalog, PySpark, Apache Iceberg, and Snowflake.

Architecture of Our Implementation

In this implementation, we will be reading a dataset and loading it as Iceberg tables into S3. We will use the AWS Glue Catalog to capture metadata, including the location of the Iceberg metadata.json file and other column-level metadata. After setting up the Iceberg tables in S3 and capturing the metadata in the Glue Catalog, we will integrate the AWS Glue Catalog with Snowflake. This integration will allow us to use Snowflake to query the data stored in S3 directly, taking advantage of Snowflake's powerful query engine to run analytics and generate insights on the data, without needing to move or duplicate the data.

This architecture allows us to store data in AWS S3 with an additional transactional layer provided by Apache Iceberg, ensuring ACID properties. The compute and storage components are decoupled, allowing users to choose their own storage and compute resources independently. This setup supports various data formats, including structured, semi-structured, and unstructured data, and can handle tasks such as data processing, real-time analytics, business intelligence, and machine learning.

Dataset

We use the New York taxi dataset, which can be downloaded from this link. This dataset includes fields capturing pick-up and drop-off dates/times, locations, trip distances, itemized fares, rate types, payment types, and passenger counts

Processing Using PySpark and Storing as Iceberg Table in S3

PySpark is a powerful distributed compute engine that, when used with Apache Iceberg, supports a wide range of features. We will read the New York taxi dataset with PySpark and write it as an Iceberg table in S3.

Step 1: Install PySpark

You can install PySpark locally using the following command:

pip install pyspark

Step 2: Read and Write Data as an Iceberg Table and Update the Glue Catalog

Below is the PySpark code to read the dataset and write it as an Iceberg table in S3, using AWS Glue Catalog for management:

from pyspark.sql import SparkSession
import os

# Set AWS environment variables
os.environ['AWS_REGION'] = 'ap-southeast-1'
os.environ['AWS_DEFAULT_REGION'] = 'ap-southeast-1'
os.environ['AWS_ACCESS_KEY_ID'] = 'XXXXXXXXX'
os.environ['AWS_SECRET_ACCESS_KEY'] = 'XXXXXXXXXX'

# Create a Spark session
spark = SparkSession.builder \
    .appName("nyc taxi") \
    .config("spark.jars.packages", 'org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.6.0,software.amazon.awssdk:bundle:2.27.7,software.amazon.awssdk:url-connection-client:2.27.7') \
    .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \
    .config("spark.sql.catalog.glue_catalog", "org.apache.iceberg.spark.SparkCatalog") \
    .config("spark.sql.catalog.glue_catalog.warehouse", "s3a://lakehouse-8765374/") \
    .config("spark.sql.catalog.glue_catalog.catalog-impl", "org.apache.iceberg.aws.glue.GlueCatalog") \
    .config("spark.sql.catalog.glue_catalog.io-impl", "org.apache.iceberg.aws.s3.S3FileIO") \
    .getOrCreate()

# Read the dataset
df = spark.read.parquet("green_tripdata_2024-01.parquet")

# Write data as Iceberg table using AWS Glue Catalog
database_name = "nyc_taxi"
table_name = "green_trip_data"

df.write.format("iceberg") \
    .mode("append") \
    .saveAsTable(f"glue_catalog.{database_name}.{table_name}")

print("Data successfully written to Iceberg table in S3 using AWS Glue Catalog.")

Configuration Details and Explanations

Let’s break down the key configurations and their importance:

• Iceberg Runtime and AWS SDK: The configuration .config("spark.jars.packages", 'org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.6.0,software.amazon.awssdk:bundle:2.27.7,software.amazon.awssdk:url-connection-client:2.27.7') is used to download the necessary Iceberg and AWS SDK packages from maven.
• Glue Catalog Configuration: The configuration .config("spark.sql.catalog.glue_catalog", "org.apache.iceberg.spark.SparkCatalog")specifies that a catalog named glue_catalog will be created and managed using Iceberg’s implementation of the SparkCatalog class, rather than Spark’s default implementation.
• The configuration .config("spark.sql.catalog.glue_catalog.catalog-impl", "org.apache.iceberg.aws.glue.GlueCatalog") points to the custom Iceberg catalog implementation class, org.apache.iceberg.aws.glue.GlueCatalog (as detailed in the Iceberg Javadoc), which in turn implements the Catalog interface (link to Catalog class).
• In this example, the glue_catalog leverages Iceberg’s SparkCatalog, but with a custom implementation provided by the org.apache.iceberg.aws.glue.GlueCatalog class.
• Warehouse Location: The configuration .config("spark.sql.catalog.glue_catalog.warehouse", "s3a://lakehouse-8765374/") sets the S3 location where the data will be stored. This tells Spark and Iceberg exactly where to store the data files in AWS S3.
• File IO Implementation: The configuration .config("spark.sql.catalog.glue_catalog.io-impl", "org.apache.iceberg.aws.s3.S3FileIO") indicates Iceberg's implementation for reading and writing data to S3.

After running the above code, the Iceberg table will be created in S3, and all the metadata details will be updated in the Glue Catalog.

So far, we have successfully created our first Iceberg table, nyc_taxi.green_trip_data. The next step is to integrate this Glue Catalog with Snowflake, allowing us to start reading the Iceberg table directly from Snowflake

Integrating Glue Catalog with Snowflake

Step 1: Create a Database in Snowflake

CREATE DATABASE lakehouse_8765374;

Step 2: Create a Schema

CREATE SCHEMA nyc_taxi;
USE lakehouse_8765374.nyc_taxi;

Step 3: Create an External Volume Object for Iceberg Table

CREATE OR REPLACE EXTERNAL VOLUME iceberg_external_volume
   STORAGE_LOCATIONS = (
         (
            NAME = 'my-s3-ap-southeast-1',
            STORAGE_PROVIDER = 'S3',
            STORAGE_BASE_URL = 's3://lakehouse-8765374/',
            STORAGE_AWS_ROLE_ARN = 'arn:aws:iam::xxxxxxx:role/sf_external_volume_role',
            STORAGE_AWS_EXTERNAL_ID = 'iceberg_table_external_id'
         )
      );

• The STORAGE_AWS_ROLE_ARN role should have the necessary permissions to access the data in S3. For more information, see Snowflake's documentation.
• Ensure that you update the trust policy of the IAM role to Snowflake’s IAM user and the external ID. These details can be obtained by running the following command:

DESC EXTERNAL VOLUME iceberg_external_volume;

Step 4: Create a Catalog Integration Object

CREATE OR REPLACE CATALOG INTEGRATION glueCatalogInt
  CATALOG_SOURCE=GLUE,
  CATALOG_NAMESPACE='nyc_taxi',
  TABLE_FORMAT=ICEBERG,
  GLUE_AWS_ROLE_ARN='arn:aws:iam::xxxxxxxx:role/sf_catalog_access_role',
  GLUE_CATALOG_ID='xxxxxxxx',
  GLUE_REGION='ap-southeast-1',
  ENABLED=TRUE;

Make sure the role arn:aws:iam::xxxxxxxx:role/sf_catalog_access_role has the necessary Glue permissions for Snowflake to access.

Refer to Snowflake’s documentation for more details.

Final Step: Create the Iceberg Table in Snowflake

CREATE ICEBERG TABLE green_trip_data
  EXTERNAL_VOLUME='iceberg_external_volume',
  CATALOG='glueCatalogInt',
  CATALOG_TABLE_NAME='green_trip_data';

If everything is configured correctly, you should be able to see and access the Iceberg table in S3 through Snowflake.

Conclusion

In this article, we implemented a lakehouse architecture using AWS S3, Glue Catalog, PySpark, Apache Iceberg, and Snowflake.

One of the key benefits of the lakehouse architecture we achieved is the decoupling of storage and compute. This allows for independent scaling, ensuring that our architecture is both cost-effective and adaptable to changing needs. Additionally, by leveraging open-source technologies like Apache Iceberg, this setup remains future-proof, with no vendor lock-in, giving us the freedom to choose any compatible compute engine. This flexibility ensures that as our data and compute needs evolve, our architecture can seamlessly adapt to new tools and technologies.

References

1. Practical Lakehouse Architecture
   O’Reilly Media
2. Apache Iceberg: The Definitive Guide
   O’Reilly Media
3. Apache Iceberg Spark Configuration Documentation
   Apache Iceberg Documentation
4. AWS Glue Catalog Integration with Apache Iceberg
   Apache Iceberg Javadoc
5. Apache Iceberg Spark Quickstart Guide
   Apache Iceberg
6. Introduction to Apache Iceberg
   Dremio Resource Guide
7. What Is a Data Lakehouse?
   Databricks Blog

Disclaimer: I used AI tools like ChatGPT just to help with grammar and phrasing. The content, ideas, and all the work in this article are 100% my own.

Introduction In this article, we will explore how to implement a simple SQLMesh project by leveraging the codebase of a dbt (Data Build Tool) project. The foundation of this article is based on a hands-on repository created by a Udemy instructor for a dbt course, available at this GitHub repo. This repository contains a comprehensive example of a dbt project, which we will use as a template to demonstrate the capabilities of SQLMesh, a next-generation data transformation tool.

Project Overview The dbt project we are using focuses on analyzing an end-to-end data engineering workflow using the Airbnb dataset in Snowflake. This project involves loading raw data into Snowflake, creating various models, and performing transformations to derive meaningful insights from the data.

Brief Explanation of the Airbnb Dataset

The Airbnb dataset used in this project consists of three primary tables: raw_listings, raw_reviews, and raw_hosts. These tables contain information about Airbnb listings, the reviews given for those listings, and details about the hosts. Understanding the relationships and significance of these tables is crucial for processing and analyzing the data.

1. raw_listings:

• Description: This table contains details about Airbnb listings, including the listing ID, URL, name, room type, minimum nights required, host ID, price, and timestamps for creation and updates.
• Significance: The listings data provides a comprehensive view of the properties available on Airbnb. It’s the central table that connects with both reviews and hosts data.

2. raw_reviews:

• Description: This table includes reviews for the Airbnb listings, containing information such as listing ID, review date, reviewer name, comments, and sentiment.
• Significance: Reviews data offers insights into the customer experience and satisfaction. It is linked to the listings data through the listing_id.

3 . raw_hosts:

• Description: This table contains information about the hosts, including host ID, name, superhost status, and timestamps for creation and updates.
• Significance: Host data provides details about the individuals or entities offering the listings. It is connected to the listings data through the host_id.

End Goal of the Project

The primary objective of this project is to analyze how full moon dates affect Airbnb ratings and reviews. By leveraging the relationships between listings, reviews, and hosts, we aim to derive meaningful insights into the impact of lunar cycles on customer feedback and satisfaction.

Steps to Implement SQLMesh Project

Setting Up SQLMesh for Snowflake

1. Initialize SQLMesh Project

• Create a folder named airbnb_sqlmesh and navigate into that folder. Run the following commands to initialize the SQLMesh project:

python -m venv .venv
source .venv/bin/activate
pip install "sqlmesh[snowflake]"
sqlmesh init snowflake

2. Project Structure After initialization, the following directories and files will be created, providing a structure to organize our SQLMesh project:

• config.yaml: Database configuration file for your project.
• models: Directory for SQL and Python models.
• audits: Directory for shared audits.
• tests: Directory for unit tests.
• macros: Directory for macros.

3. Project Configuration : Define SQLMesh project configurations in the config.yaml file. This file contains database connection details and model defaults:

gateways:
  snowflake_conn:
    connection:
      type: snowflake
      user: <username>
      password: <password>
      account: <account>
      warehouse: <warehouse>
      database: AIRBNB
      role: <role>
default_gateway: snowflake_conn

model_defaults:
  dialect: snowflake
  start: 2009-01-01

4. Loading Raw Data Use the following SQL scripts to load the raw data into Snowflake tables:

CREATE DATABASE IF NOT EXISTS AIRBNB;
CREATE SCHEMA IF NOT EXISTS AIRBNB.RAW;

USE DATABASE airbnb;
USE SCHEMA RAW;

-- Create and load raw_listings table
CREATE OR REPLACE TABLE raw_listings (
  id integer,
  listing_url string,
  name string,
  room_type string,
  minimum_nights integer,
  host_id integer,
  price string,
  created_at datetime,
  updated_at datetime
);

COPY INTO raw_listings FROM 's3://dbtlearn/listings.csv'
  FILE_FORMAT = (type = 'CSV' FIELD_OPTIONALLY_ENCLOSED_BY = '"');

-- Create and load raw_reviews table
CREATE OR REPLACE TABLE raw_reviews (
  listing_id integer,
  date datetime,
  reviewer_name string,
  comments string,
  sentiment string
);

COPY INTO raw_reviews FROM 's3://dbtlearn/reviews.csv'
  FILE_FORMAT = (type = 'CSV' FIELD_OPTIONALLY ENCL_ENCLOSED_BY = '"');

-- Create and load raw_hosts table
CREATE OR REPLACE TABLE raw_hosts (
  id integer,
  name string,
  is_superhost string,
  created_at datetime,
  updated_at datetime
);

COPY INTO raw_hosts FROM 's3://dbtlearn/hosts.csv'
  FILE_FORMAT = (type = 'CSV' FIELD_OPTIONALLY_ENCLOSED_BY = '"');

Creating Models

1. Source Models We create three models src_hosts.sql, src_listings.sql, and src_reviews.sql on top of the raw data tables inside the models/source folder:

The source models in this project are designed to standardize and prepare raw data from the Airbnb dataset for further processing and analysis. They act as an intermediary layer, transforming raw data into a more usable format by renaming columns, selecting relevant fields, and standardizing the data. All source models are materialized as views, ensuring that the data is clean, structured, and ready for subsequent transformations and analysis in the dimension and fact models.

MODEL (
  name SOURCE.SRC_HOSTS,
  kind view
);

WITH raw_hosts AS (
  SELECT * FROM RAW.RAW_HOSTS
)
SELECT
  id AS host_id,
  name AS host_name,
  is_superhost,
  created_at,
  updated_at
FROM raw_hosts;

MODEL (
  name SOURCE.SRC_LISTINGS,
  kind view
);

WITH raw_listings AS (
  SELECT * FROM RAW.RAW_LISTINGS
)
SELECT
  id AS listing_id,
  name AS listing_name,
  listing_url,
  room_type,
  minimum_nights,
  host_id,
  price AS price_str,
  created_at,
  updated_at
FROM raw_listings;

MODEL (
  name SOURCE.SRC_REVIEWS,
  kind view
);

WITH raw_reviews AS (
  SELECT * FROM raw.raw_reviews
)
SELECT
  listing_id,
  date AS review_date,
  reviewer_name,
  comments AS review_text,
  sentiment AS review_sentiment
FROM raw_reviews;

2. Dimension Models Create dimension models as views inside the models/dim folder:

The dimension models clean, enrich, and combine the source data for detailed analysis. They handle null values, format fields, and merge related datasets. Models dim.dim_hosts_cleansed and dim.dim_listings_cleansed are materialized as views, while dim.dim_listings_w_hosts is materialized as a full table, reloaded entirely each time.

MODEL (
  name dim.dim_hosts_cleansed,
  kind view
);

WITH src_hosts AS (
  SELECT * FROM SOURCE.SRC_HOSTS
)
SELECT
  host_id,
  NVL(host_name, 'Anonymous') AS host_name,
  is_superhost,
  created_at,
  updated_at
FROM src_hosts;

MODEL (
  name dim.dim_listings_cleansed,
  kind view
);

WITH src_listings AS (
  SELECT * FROM SOURCE.SRC_LISTINGS
)
SELECT
  listing_id,
  listing_name,
  room_type,
  CASE WHEN minimum_nights = 0 THEN 1 ELSE minimum_nights END AS minimum_nights,
  host_id,
  REPLACE(price_str, '$')::NUMBER(10, 2) AS price,
  created_at,
  updated_at
FROM src_listings;

MODEL (
  name dim.dim_listings_w_hosts,
  kind full
);

WITH l AS (
  SELECT * FROM dim.dim_listings_cleansed
),
h AS (
  SELECT * FROM dim.dim_hosts_cleansed
)
SELECT
  l.listing_id,
  l.listing_name,
  l.room_type,
  l.minimum_nights,
  l.price,
  l.host_id,
  h.host_name,
  h.is_superhost AS host_is_superhost,
  l.created_at,
  GREATEST(l.updated_at, h.updated_at) AS updated_at
FROM l
LEFT JOIN h ON (h.host_id = l.host_id);

3 .Fact Models Create incremental fact models inside the models/fct folder:

The fact model, fct.reviews, processes and aggregates review data from the source models. It is an incremental model, meaning it only loads new data, which drastically decreases computational resources required for each model run. In the model, the @start and @end_date are SQLMesh macros that are rendered during the sqlmesh plan or run with the appropriate start and end date for the run. Additionally, the @GENERATE_SURROGATE_KEY macro is used to generate an MD5 hash for the given input columns, creating a surrogate key for the data.

MODEL (
  name fct.reviews,
  kind INCREMENTAL_BY_TIME_RANGE (
    time_column review_date
  )
);

WITH src_reviews AS (
  SELECT * FROM SOURCE.SRC_REVIEWS
)
SELECT
  @GENERATE_SURROGATE_KEY(listing_id, review_date, reviewer_name, review_text) AS review_id,
  *
FROM src_reviews
WHERE review_text IS NOT NULL
  AND review_date BETWEEN @start_date AND @end_date;

4. Business Layer Models: Moving on to the business layer, we need to analyze how full moon dates affect review comments. To do this, we require a table containing the dates of the full moons. This can be achieved using a seed file in SQLMesh. Download the seed file from the provided link and place it in the seeds folder of the SQLMesh project. Then, create a seed model named full_moon_dates_seed.sql inside the models folder.

MODEL (
  name seed.full_moon_dates,
  kind SEED (
    path '../seeds/seed_data.csv'
  )
);

For the final model, create a folder named models/mart and add the following model inside it

MODEL (
  name mart.mart_fullmoon_reviews,
  kind full
);

WITH fct_reviews AS (
    SELECT * FROM fct.reviews
),
full_moon_dates AS (
    SELECT * FROM seed.full_moon_dates
)

SELECT
  r.*,
  CASE
    WHEN fm.full_moon_date IS NULL THEN 'not full moon'
    ELSE 'full moon'
  END AS is_full_moon
FROM
  fct_reviews
  r
  LEFT JOIN full_moon_dates
  fm
  ON (TO_DATE(r.review_date) = DATEADD(DAY, 1, fm.full_moon_date))

Once all the models are created, the project structure should look like this:

Running SQLMesh Plan

Execute the SQLMesh plan command to check and deploy the models:

sqlmesh plan dev

Specify the backfill start and end dates when prompted. After confirming the backfills, the models will be deployed to the development environment.

Once the model is executed successfully , run the sqlmesh plan against the product

sqlmesh plan prod

SQLMesh will not reprocess all the data as it did for the development environment. Instead, it reuses the models created in the development environment. For more information on SQLMesh virtual environments, refer to this link.

Disclaimer: I used AI tools like ChatGPT just to help with grammar and phrasing. The content, ideas, and all the work in this article are 100% my own.