Use SQLMesh to streamline pipelines

Introduces how SQLMesh simplifies managing RisingWave streaming SQL pipelines.

SQLMesh provides a structured and maintainable approach to managing end-to-end streaming pipelines. As the control plane for RisingWave, it simplifies the workflow by enabling local definition of transformations, planning and reviewing changes before deployment, and running optional audits to ensure data quality.

This guide explains how to use RisingWave for streaming data and SQLMesh for managing transformations to aggregate website click events. Each event contains a timestamp, event_type, and value.

It aims to calculate a rolling 5-minute count and sum of value for each event_type using a tumbling window. Then, the logic will be modified to calculate an average, showcasing how SQLMesh handles the change.

Set up RisingWave and ingest sample data

Begin by starting RisingWave and manually inserting sample data into a base table.

1. Start RisingWave via Docker.

   docker run -it --pull=always -p 4566:4566 -p 5691:5691 risingwavelabs/risingwave:latest single_node
   

   The message of RisingWave standalone mode is ready means that RisingWave has started successfully.

2. Connect via psql (or another Postgres-compatible SQL client) and prepare the data.

   # Use localhost when connecting from your machine to the mapped port
   psql -h localhost -p 4566 -d dev -U root
   

   Inside psql, run the following:

   -- Create the source table to hold incoming data
   CREATE TABLE click_events (
       event_id INT PRIMARY KEY,
       event_type VARCHAR,
       event_value INT,
       event_timestamp TIMESTAMPTZ
   );
   
   -- Insert seven rows of sample data
   INSERT INTO click_events (event_id, event_type, event_value, event_timestamp) VALUES
   (1, 'page_view', 10, '2023-10-27 10:00:30+00'),
   (2, 'add_to_cart', 150, '2023-10-27 10:01:15+00'),
   (3, 'page_view', 12, '2023-10-27 10:02:05+00'),
   (4, 'page_view', 11, '2023-10-27 10:05:40+00'),
   (5, 'purchase', 250, '2023-10-27 10:06:20+00'),
   (6, 'page_view', 15, '2023-10-27 10:11:00+00'),
   (7, 'add_to_cart', 99, '2023-10-27 10:12:30+00');
   
   -- Optional: Verify inserts
   SELECT COUNT(*) FROM click_events;
   
   -- Exit psql
   \q
   

Set up SQLMesh project

Set up the SQLMesh environment to manage our RisingWave transformations.

1. Create a project directory and Python virtual environment.

   mkdir sqlmesh-risingwave-demo
   cd sqlmesh-risingwave-demo
   python3 -m venv venv
   source venv/bin/activate # Adjust activation for your OS/shell
   

2. Install the SQLMesh RisingWave adapter:

   pip install "sqlmesh[risingwave]"
   

3. Initialize the SQLMesh Project specifically for RisingWave:

   sqlmesh init risingwave
   

   This creates necessary folders and configuration files.

4. Configure the connection in config.yaml file (created by init). Ensure the following matches or update your config.yaml.

   gateways:
   risingwave: # Gateway name used by default
       connection:
       type: risingwave
       host: localhost
       user: root
       port: 4566 # Match the Docker port mapping
       database: dev
   
   default_gateway: risingwave
   
   model_defaults:
   dialect: risingwave
   # start: <YYYY-MM-DD> # Optional: Default start date for backfills if needed
   

Define and deploy a streaming query (v1)

Create the first transformation model and deploy it.

1. Create the SQLMesh model file models/event_summary_tumbling.sql:

   -- models/event_summary_tumbling.sql
   MODEL (
     name reporting.event_summary_tumbling,
     kind VIEW (materialized = true), -- This makes it a RisingWave MV
     owner data_team,
     description 'Summarizes event counts and total value in 5-minute tumbling windows.'
   );
   
   SELECT
       window_start,
       window_end,
       event_type,
       COUNT(*) AS event_count,
       SUM(event_value) AS total_value
   FROM TUMBLE(
       click_events,         -- Read from the base table
       event_timestamp,            -- Time column for windowing
       INTERVAL '5 minutes'        -- Tumbling window size
   )
   GROUP BY
       window_start,
       window_end,
       event_type;
   

2. Plan and apply the model using SQLMesh:

   sqlmesh plan
   

   SQLMesh will detect the new model (reporting.event_summary_tumbling) and show a plan to create it in the prod environment (which maps to a schema like sqlmesh_prod or prod in RisingWave, often including the model name itself, e.g., sqlmesh__reporting). It will also detect the default models created by init; you can choose to apply changes only for your new model if desired, or apply all.

   # Example Output Snippet:
   # Summary of differences:
   # Models:
   # └── Added:
   #     └── reporting.event_summary_tumbling
   #     └── sqlmesh_example.full_model (if applying init models)
   #     └── ... (other init models)
   #
   # Apply - Apply the plan. [y/n]: y  <-- Type 'y' and press Enter
   

   SQLMesh executes the CREATE MATERIALIZED VIEW statement(s) in RisingWave.

3. Verify the MV creation and content via psql .

   The exact schema and MV name includes a hash for versioning. You can use SHOW to find it. Since reporting is used as model schema, the materialized view should be in the sqlmesh__reporting schema.

   show materialized views from sqlmesh__reporting;
   -- Result below
                       Name                      
   -----------------------------------------------
   reporting__event_summary_tumbling__3036268106
   (1 row)
   

   You should see the aggregated results similar to below.

   

Modify the MV schema with SQLMesh (v2)

Now we can change the aggregation logic and see how SQLMesh manages the update safely.

1. Modify the model file models/event_summary_tumbling.sql to calculate the average value and distinct count:

   -- models/event_summary_tumbling.sql
   MODEL (
   name reporting.event_summary_tumbling, -- Keep the same name
   kind VIEW (materialized true),
   owner data_team,
   -- Update description to reflect change
   description 'Calculates average event value and distinct event count in 5-minute tumbling windows.'
   );
   
   SELECT
       window_start,
       window_end,
       event_type,
       COUNT(DISTINCT event_id) AS distinct_event_count, -- Changed aggregation
       AVG(event_value) AS average_value          -- Changed aggregation
   FROM TUMBLE(
       click_events,
       event_timestamp,
       INTERVAL '5 minutes'
   )
   GROUP BY
       window_start,
       window_end,
       event_type;
   

2. Run sqlmesh plan again to see the impact.

   SQLMesh detects the change in the definition of reporting.event_summary_tumbling. Because the logic changed, it plans to:

   • Create a new version of the Materialized View in RisingWave with a different physical name (a new hash suffix).
   • It does not immediately drop the old version, allowing for validation or zero-downtime promotion strategies (though in this simple apply, the old one might eventually be cleaned up depending on settings).
   

3. Verify the new MV version via psql:

   -- List MVs again in the schema
   SHOW MATERIALIZED VIEWS FROM sqlmesh__reporting; -- Replace schema if needed
   -- You should now see TWO MVs related to event_summary_tumbling,
   -- one with the old hash and one with a NEW hash.
   
   -- Query the NEWER MV version (find its exact name)
   SELECT * FROM sqlmesh__reporting."reporting__event_summary_tumbling__<new_hash>" ORDER BY window_start, event_type;
   

   The results should show the distinct_event_count and average_value columns, reflecting the v2 logic. The original v1 MV still exists momentarily. This demonstrates SQLMesh’s safe, versioned deployment approach.

Integrate sink and source management

This guide manually created the click_events table in RisingWave for simplicity. However, SQLMesh provides pre- and post-statements within model definitions, allowing you to embed DDL statements that run before or after the main model logic executes.

For example, you could place a CREATE SOURCE IF NOT EXISTS ... statement in pre-statement (that is, a SQL query prior to the SELECT statement), and / or a CREATE SINK IF NOT EXISTS ... in a post-statement. This approach keeps the setup/teardown logic closer to the relevant transformation step within your SQLMesh project.

SQLMesh’s core focus is managing the transformation logic (the SELECT statements) and dependencies across your entire pipeline (often multiple models). While pre- and post-statements offer integration points, managing complex CREATE SOURCE/SINK statements with intricate connector configurations might still be cleaner handled outside the model definitions (e.g., separate setup scripts or specialized infrastructure tools) to maintain clarity.

Additional resources

• SQLMesh Documentation

• Learn more about SQLMesh