Set up Insights on Databricks Lakebase

October 6, 2026

This section describes how to set up Insights Data Access on Databricks, using a Databricks Lakehouse volume for file storage and Databricks Lakebase (PostgreSQL) as the query database.

Tip For information on how to set up Insights Data Access on Amazon Web Services, go to Set up Insights on AWS. For information on how to set up Insights Data Access on the Google Cloud Platform, go to Set up Insights on GCP.

Prerequisites

You have the following:

Collibra Platform 5.7 or newer.
License for Collibra Insights.
Software for working with Parquet files.

A Databricks workspace with Unity Catalog enabled, and a principal with the following permissions:

Area	Permission	Purpose
Catalog and volume	CREATE CATALOG or USE CATALOG, USE SCHEMA, READ VOLUME, WRITE VOLUME	Creating the landing volume and uploading the Insights export.
Secrets	CREATE_SCOPE, WRITE and READ on the secret scope	Storing database and API credentials used by the notebook.
Compute and jobs	CAN_USE on a cluster, CREATE_JOB, CAN_START or CAN_MANAGE on a job	Running and scheduling the ETL notebook.
Lakebase (Postgres)	CONNECT, USAGE and CREATE on the target schema, INSERT, UPDATE, TRUNCATE and SELECT on the Insights tables	Loading and querying the Insights data in Lakebase.

Steps

Download a data snapshot from your Collibra environment
Set up Databricks Lakehouse and Lakebase and test the workflow
Prepare and run the necessary scripts for the ETL process
Automate the workflow (optional)

Step 1: Download a data snapshot from your Collibra environment

Enter the following URL in your browser:
<your-Collibra-environment-URL>/rest/2.0/reporting/insights/directDownload?snapshotDate=<snapshot_date>&format=zip
Tip <snapshot date> is the date from when you want the data, formatted as YYYY-MM-DD, for example, 2023-09-29. Ensure that the date you enter is within the last 31 days or is the last day of a month.
A ZIP file of the data from your Collibra environment, for the specified date, is downloaded to your hard disk.
Extract the ZIP files on your local computer.
A folder with the name of the ZIP file is created.

Step 2: Set up Databricks Lakehouse and Lakebase and test the workflow

Extract Data Insights Parquet files from Collibra cloud using the following API call: <your-Collibra-environment-URL>/rest/2.0/reporting/insights/directDownload?snapshotDate=<snapshot_date>&format=zip
The zip file downloads to your desktop with name format of insights_YYYY-MM-DD.zip.
Once the download is complete, extract the zip file in your local desktop.
Create a Lakehouse volume and upload the Parquet files to the Databricks Lakehouse volume:
1. Log in to the Databricks console and click Create > Create a Catalog.
2. Select the appropriate options for the new catalog and click Create.
3. Select the catalog in the left panel and click Create Schema.
4. Open the schema and click Create > Volume.
5. Open the volume and click Upload to this volume.
6. In the pop-up, click Browse > Select folder.
7. Confirm the data is readable by running the following command in a notebook cell or a SQL editor: spark.read.parquet('/Volumes/main/collibra_insights/landing/asset/').limit(5).display()

Create a PgSql lakebase and other required objects:

In the left pane, click Compute.
On the Lakebase tab, click Create database project.
Enter the Display name and click Create.
Click on the Provisioned and Autoscaling options to view the instance.
Click on the new project to open it in a new window.
Click Connect and note the JDBC URL/connection string. This is needed later to connect to the database and load the transformed data (Parquet files) into the appropriate tables.
Select the appropriate Branch and Database.
In SQL Editor, run the following queries to create the required database and schema.
CREATE DATABASE "collibra-rpt"; CREATE SCHEMA IF NOT EXISTS insights;

In SQL Editor, create a Lakebase role and grant the necessary permissions.

Copy

--Create role with password
CREATE ROLE collibra_loader WITH LOGIN PASSWORD '<choose-a-strong-password>';

-- Grant permission to create schemas within the target database
GRANT CREATE ON DATABASE "collibra-rpt" TO collibra_loader;

-- 1. Give the user access to see and use the schema
GRANT USAGE ON SCHEMA insightsvis TO collibra_loader;

-- 2. Give the user permission to create/overwrite tables inside the schema
GRANT CREATE ON SCHEMA insightsvis TO collibra_loader;

Create a scope and store the credentials as secrets. These will be run in a separate Python notebook.

Copy

from databricks.sdk import WorkspaceClient
from databricks.sdk import WorkspaceClient
from datetime import datetime, timedelta
# 1. Create the scope
w = WorkspaceClient()
w.secrets.create_scope(scope="your_scope_name")
print("Secret scope 'lakebase' created successfully.")
# 2. Populate the secrets
w.secrets.put_secret(scope="your_scope_name", key="user", string_value="---------")
w.secrets.put_secret(scope="your_scope_name", key="password", string_value="-------")
print("Secrets added successfully.")
print("�� Initializing Connection Details, Pipeline Metadata, and Security Scopes...")
# Initialize the Databricks Workspace Client
w = WorkspaceClient()
print("�� Injecting new OAuth integration credentials into 'your_scope_name' scope...")
# 1. Populate your Collibra Client ID
w.secrets.put_secret(
    scope="your_scope_name", 
    key="collibra-client-id", 
    string_value="urn%3Asys%3Aenv%3Ae85d28dc-e88f-40ae-867d-37ff0e002cad%3Ai%3Auf5ctd" 
)
# 2. Populate your Collibra Client Secret
w.secrets.put_secret(
    scope="your_scope_name", 
    key="collibra-client-secret", 
    string_value="X4DEqpuBoMKF2CD-2ng_5lw37Ff3kfdjNTjDbQMqC7FvQPet5dOpHGF3NxTgSgS3"  
)
print("✅ Secrets added successfully. Delete this cell or clear your keys before saving!")

Step 3: Prepare and run the necessary scripts for the ETL process

Note The steps above are run in the SQL Editor and the steps below are run in Python notebook.

Create a Python notebook in your workspace.
Select the appropriate compute resource to run the code.

Define the connection parameters in the third cell. Numbering in this procedure begins with Cell 3. Cells 1 and 2 are reserved for automation logic, which is explained in Step 5: Automate the workflow.

Copy

# Cell 3 — connection parameters
import os
from datetime import datetime, timedelta

print("�� Initializing Connection Details and Pipeline Metadata...")

# 1. Clean serverless database endpoint domain
LAKEBASE_HOST = "--------.cloud.databricks.com"
LAKEBASE_PORT = "5432"
LAKEBASE_DB   = "-------"

# �� AUTOMATED DATE CALCULATOR
yesterday     = datetime.now() - timedelta(days=1)
SNAPSHOT_DATE = yesterday.strftime("%Y-%m-%d")

# 2. Workspace storage and table arrays for your pipeline
VOLUME_ROOT = f"/Volumes/subhash/hs_schema/collibra_datainsights/Extractedfiles/{SNAPSHOT_DATE}/"

# 3. Core table arrays for your pipeline execution loop
TABLES = [
    "asset", "asset_tag", "attribute", "community",
    "complex_relation", "domain", "relation", "responsibility", "view_events"
]

# 4. Dynamically pull the Database Username & Password credentials from your secret scope
DB_USER     = dbutils.secrets.get(scope="your_scope_name", key="user")
DB_PASSWORD = dbutils.secrets.get(scope="your_scope_name", key="password")

# 5. Construct standard JDBC string used by Spark
jdbc_url = f"jdbc:postgresql://{LAKEBASE_HOST}:{LAKEBASE_PORT}/{LAKEBASE_DB}"
print(f"�� Database target mapped to: {jdbc_url}")
print(f"�� Ingestion Target Path: {VOLUME_ROOT}")
print("✅ Cell 1 configuration parameters and secure credentials loaded successfully.")

Test the JDBC connection with the created secrets in the fourth cell. Data is written to relational tables based on the success of this step. Run the above two cells to make sure that the connection is established successfully. Resolve any connection issues at this stage. When you add a new cell, make sure to run all the cells one by one, starting from cell 3, to test continuous execution among cells.

Copy

# Cell 4 — smoke test (Cleaned up for Serverless consistency)
check = (spark.read
    .format("postgresql")
    .option("host", "-------.cloud.databricks.com")
    .option("port", "5432")
    .option("database", "--------")
    .option("user", dbutils.secrets.get(scope="your_scope_name", key="user"))
    .option("password", dbutils.secrets.get(scope="your_scope_name", key="password"))
    .option("ssl", "true")
    .option("sslmode", "require")
    .option("dbtable", "(SELECT version() AS v) AS t")
    .load())
display(check)

Create the first table in the fifth cell.

Copy

# Cell 5 — per-table loader
def load_table(table_name: str, mode: str = "overwrite") -> int:
    src = f"{VOLUME_ROOT}/{table_name}/"
    
    try:
        df  = spark.read.parquet(src)
        n   = df.count()
        
        if n == 0:
            print(f"Skipped {table_name}: Source directory is empty.")
            return 0
            
        (df.write
            .format("postgresql")               
            .option("host", "-----------cloud.databricks.com")
            .option("port", "5432")
            .option("database", "----------") # Fixed target DB context
            .option("user", dbutils.secrets.get(scope="your_scope_name", key="user"))   
            .option("password", dbutils.secrets.get(scope="your_scope_name", key="password"))   
            .option("dbtable", f"insights.{table_name}")
            .option("batchsize", 5000)
            .option("truncate", "true")        
            .option("isolationLevel", "NONE")  
            .mode(mode)
            .save())
        print(f"Loaded {n:>10,} rows into insights.{table_name}") # Fixed log schema name
        return n
        
    except Exception as e:
        print(f"❌ Error loading {table_name}: {str(e)}")
        return -1

Create and load the remaining tables in the sixth cell.

Copy

# Cell 6 — load remaining tables
totals = {}

print("Starting batch ingestion pipeline...\n")
for t in TABLES:
    # Runs each table individually and stores the resulting row count
    totals[t] = load_table(t)

print("\n==============================")
print("Load summary:")
print("==============================")
for t, n in totals.items():
    if n == -1:
        print(f"  {t:<18} ❌ FAILED")
    else:
        print(f"  {t:<18} {n:>10,} rows")

Once all the tables are created, it is advisable to create indexes for the appropriate primary key columns by running the query below from the Lakebase SQL Editor. This will execute queries on the Lakebase faster.

Copy

-- ===========================================================================
-- insights.asset
-- Joined constantly via (asset_id, snapshot_date); filtered by domain_id
-- and asset_type_id in process_register / privacy_risk_readiness.
-- ===========================================================================
CREATE INDEX IF NOT EXISTS idx_asset_id_snapshot
    ON insights.asset (asset_id, snapshot_date);

CREATE INDEX IF NOT EXISTS idx_asset_domain_snapshot
    ON insights.asset (domain_id, snapshot_date);

CREATE INDEX IF NOT EXISTS idx_asset_snapshot_type
    ON insights.asset (snapshot_date, asset_type_id);

-- ===========================================================================
-- insights.domain
-- Joined via (domain_id, snapshot_date); also the source of the
-- MAX(snapshot_date) subquery used in WHERE clauses of ALL THREE views;
-- also filtered by domain_type_id in privacy_risk_readiness.
-- ===========================================================================
CREATE INDEX IF NOT EXISTS idx_domain_id_snapshot
    ON insights.domain (domain_id, snapshot_date);

CREATE INDEX IF NOT EXISTS idx_domain_snapshot_date
    ON insights.domain (snapshot_date);        -- speeds up MAX(snapshot_date) in all 3 views

CREATE INDEX IF NOT EXISTS idx_domain_type_snapshot
    ON insights.domain (domain_type_id, snapshot_date);

CREATE INDEX IF NOT EXISTS idx_domain_community_snapshot
    ON insights.domain (community_id, snapshot_date);

-- ===========================================================================
-- insights.community  (used in data_maturity)
-- ===========================================================================
CREATE INDEX IF NOT EXISTS idx_community_id_snapshot
    ON insights.community (community_id, snapshot_date);

-- ===========================================================================
-- insights.relation
-- Joined via BOTH asset_id1 and asset_id2 across the three views, almost
-- always filtered by relation_type_id / role_or_corole in the same predicate.
-- ===========================================================================
CREATE INDEX IF NOT EXISTS idx_relation_asset1_snapshot_type
    ON insights.relation (asset_id1, snapshot_date, relation_type_id);

CREATE INDEX IF NOT EXISTS idx_relation_asset2_snapshot_type
    ON insights.relation (asset_id2, snapshot_date, relation_type_id);

CREATE INDEX IF NOT EXISTS idx_relation_asset1_snapshot_role
    ON insights.relation (asset_id1, snapshot_date, role_or_corole);  -- process_register, data_maturity role filters

-- ===========================================================================
-- insights.attribute
-- Always filtered by (asset_id, snapshot_date) + attribute_type_name
-- (data_maturity, process_register) or attribute_type_id (privacy_risk_readiness).
-- ===========================================================================
CREATE INDEX IF NOT EXISTS idx_attribute_asset_snapshot_typename
    ON insights.attribute (asset_id, snapshot_date, attribute_type_name);

CREATE INDEX IF NOT EXISTS idx_attribute_asset_snapshot_typeid
    ON insights.attribute (asset_id, snapshot_date, attribute_type_id);

-- ===========================================================================
-- insights.responsibility
-- Filtered by (asset_id, snapshot_date) + role_id
-- (Owner / Business Steward / Privacy Steward / Data Steward in privacy_risk_readiness;
--  generic role lookup in data_maturity).
-- ===========================================================================
CREATE INDEX IF NOT EXISTS idx_responsibility_asset_snapshot_role
    ON insights.responsibility (asset_id, snapshot_date, role_id);

From any Postgres client (Eg: Dbeaver), connect to collibra_rpt and confirm that all tables exist in the seventh cell.

Compare the Lakebase row counts to the Parquet row counts in the Lakebase SQL Editor. The following query returns one row per table:

Copy

SET search_path TO insights;


SELECT 'asset'             AS table_name, COUNT(*) FROM insights.asset            UNION ALL
SELECT 'asset_tag',                       COUNT(*) FROM insights.asset_tag        UNION ALL
SELECT 'attribute',                       COUNT(*) FROM insights.attribute        UNION ALL
SELECT 'community',                       COUNT(*) FROM insights.community        UNION ALL
SELECT 'complex_relation',                COUNT(*) FROM insights.complex_relation UNION ALL

Insert a cell below the seventh cell and add this code for verification:

Copy

# Cell 7 — Automated Post-Load Row Count Verification Gate

print("�� Starting Post-Load Row Count Validation...")

# 1. Fetch current row counts directly via standard Spark JDBC connection
validation_errors = []

# Safety Check: Verify variables from Cell 1 are loaded properly in memory
try:
    jdbc_url
except NameError:
    raise NameError("❌ CONFIGURATION ERROR: Connection parameters from Cell 1 are not in memory. Please run Cell 1 first.")
   
   for table in TABLES:
    try:
        # Queries your target postgres database tables using standard, robust JDBC driver format
        count_df = (spark.read
            .format("jdbc")
            .option("url", jdbc_url)  # �� Uses your verified connection string from Cell 1
            .option("dbtable", f"insights.{table}")
            .option("user", dbutils.secrets.get(scope="your_scope_name", key="user"))
            .option("password", dbutils.secrets.get(scope="your_scope_name", key="password"))
            .load())

# �� CRITICAL GUARD: Check if the table is completely empty
        if row_count == 0:
            validation_errors.append(f"❌ DATA INTEGRITY BREACH: Table 'insights.{table}' is empty!")

except Exception as table_err:
        validation_errors.append(f"❌ CONNECTIVITY ERROR: Could not read table 'insights.{table}': {str(table_err)}")

# 2. Evaluation Phase
print("\n==============================================")
print("Validation Results Summary:")
print("==============================================")
if validation_errors:
    print(f"�� INTEGRITY CHECK FAILED: {len(validation_errors)} error(s) discovered during database analysis:")
    for error in validation_errors:
        print(error)
    print("\n�� Halting pipeline execution with fatal failure status.")
    # This intentionally crashes the cell execution, forcing the Databricks scheduler to fail the job run
    raise ValueError("Pipeline validation failed: One or more database target tables are empty.")
else:
    print("✅ INTEGRITY CHECK PASSED: All target ingestion tables verified active with valid rows.")
    print("�� Daily pipeline completed successfully!")

Step 4: Automate the workflow (optional)

Add another cell at the top of the notebook and enter the code below to generate the OAuth token. Make adjustments according to your connection properties. This cell now becomes the first cell in the notebook.

Copy

# Cell 1 — OAuth Token Generation & Automated Date Parameter
import requests
from datetime import datetime, timedelta

print("�� Initializing Session and Requesting Dynamic OAuth Access Token...")

# 1. Automated Snapshot Date Calculation (Targets yesterday's run)
yesterday     = datetime.now() - timedelta(days=1)
SNAPSHOT_DATE = yesterday.strftime("%Y-%m-%d")

# 2. Base API URL Parameters for Auth
COLLIBRA_URL  = "https://dg-qa-pb4.collibra.com/"  
TOKEN_URL     = f"{COLLIBRA_URL}/rest/oauth/v2/token"

# 3. Fetch Client Credentials securely from Databricks secret scope
CLIENT_ID     = dbutils.secrets.get(scope="your_scope_name", key="collibra-client-id")
CLIENT_SECRET = dbutils.secrets.get(scope="your_scope_name", key="collibra-client-secret")

# 4. Execute Dynamic OAuth Client Credentials Grant Request
# Form body payload now only requires the grant type descriptor
payload = {
    "grant_type": "client_credentials"
}

headers = {
    "Content-Type": "application/x-www-form-urlencoded"
}

try:
    # �� FIXED: credentials are now passed via basic auth header instead of body
    response = requests.post(
        TOKEN_URL, 
        data=payload, 
        headers=headers,
        auth=(CLIENT_ID, CLIENT_SECRET) 
    )
    
    if response.status_code == 200:
        token_data = response.json()
        # Saves token to RAM memory for the next cells to use
        API_TOKEN = token_data["access_token"] 
        print(f"✅ OAuth Success: Access Token generated for target date: {SNAPSHOT_DATE}")
    else:
        print(f"❌ OAuth Authentication Failed: HTTP {response.status_code} - {response.text}")
        raise requests.exceptions.HTTPError("Authentication failure: token could not be issued.")
except Exception as e:
    print(f"�� Failed to establish connection with Auth Server: {str(e)}")
    raise e

Add another cell below the OAuth token code and enter the code below. This code connects to the Insights through the Insights API, downloads the zip file to the set location, unzips the zip file, and extracts the Parquet files to the set directory structure and location.

Copy

# Cell 2 — Downloading Snapshot and Unpacking
import os
import requests
import zipfile
from datetime import datetime, timedelta

print("�� Downloading Collibra Data Insights Snapshot...")

# 1. Calculate Target Date (Today's date - 1 day)
today = datetime.now()
api_date_obj = today - timedelta(days=1)
API_SNAPSHOT_DATE = api_date_obj.strftime("%Y-%m-%d")

# 2. Dynamic storage paths matching the actual downloaded yesterday file
VOLUME_ROOT   = f"/Volumes/subhash/hs_schema/collibra_datainsights/Extractedfiles/{API_SNAPSHOT_DATE}/"
ZIP_FILE_PATH = f"/Volumes/subhash/hs_schema/collibra_datainsights/zipfiles/{API_SNAPSHOT_DATE}.zip"
EXTRACT_DIR   = VOLUME_ROOT

print("�� Initializing Storage Paths and Target Database Metadata...")
print(f"�� Calculated API snapshot date (Yesterday): {API_SNAPSHOT_DATE}")

# 3. Construct API Endpoint URL using COLLIBRA_URL from Cell 1
download_url = f"{COLLIBRA_URL.rstrip('/')}/rest/2.0/reporting/insights/directDownload?snapshotDate={API_SNAPSHOT_DATE}&format=zip"

# 4. Construct Authorization Headers using the token generated in Cell 1
download_headers = {
    "Authorization": f"Bearer {API_TOKEN}"
}

# 5. Create local directories if they do not exist
os.makedirs(os.path.dirname(ZIP_FILE_PATH), exist_ok=True)
os.makedirs(EXTRACT_DIR, exist_ok=True)

# 6. Download the ZIP file using Bearer Auth
try:
    print(f"�� Requesting download from Collibra...")
    response = requests.get(download_url, headers=download_headers, stream=True)
    response.raise_for_status()
    
    # Write the stream data to your ZIP file path
    with open(ZIP_FILE_PATH, "wb") as file:
        for chunk in response.iter_content(chunk_size=8192):
            if chunk:
                file.write(chunk)
                
    # Requested output confirmation statement
    print(f"✅ Snapshot with date {API_SNAPSHOT_DATE} has been successfully downloaded.")

    # ========================================================
    # 7. UNPACKING THE ZIP FILE TO EXTRACTEDFILES DIRECTORY
    # ========================================================
    print(f"�� Unpacking zip archive to {EXTRACT_DIR}...")
    with zipfile.ZipFile(ZIP_FILE_PATH, 'r') as zip_ref:
        zip_ref.extractall(EXTRACT_DIR)
    print(f"✨ Successfully extracted all contents to Databricks Volume folder.")

except requests.exceptions.RequestException as e:
    print(f"❌ Failed to download snapshot: {e}")
except zipfile.BadZipFile:
    print(f"❌ Failed to unpack file: The downloaded file is not a valid ZIP archive or is corrupted.")
except Exception as e:
    print(f"�� An unexpected error occurred during processing: {e}")

Once you create the views associated with the Tableau workbook, add another cell with the code below.

Copy

# Databricks notebook source
# MAGIC %md
# MAGIC ## Refresh Privacy/Governance Materialized Views
# MAGIC Refreshes, in order, 15 minutes apart:
# MAGIC 1. insights.data_maturity
# MAGIC 2. insights.process_register
# MAGIC 3. privacy_risk_readiness
# MAGIC
# MAGIC Adjust the wait time in the `WAIT_MINUTES` variable below if 15 min needs to change.

# COMMAND ----------

# MAGIC %pip install psycopg2-binary

# COMMAND ----------

import time
import datetime
import base64
import psycopg2
from databricks.sdk import WorkspaceClient

# ---------------------------------------------------------------------------
# CONFIG: Connection details
# user/password are pulled from the 'your_scope_name' secret scope you created.
# host/port/dbname are NOT currently stored as secrets in that scope, so fill
# them in below (or add them as secrets too and swap in get_secret calls).
# ---------------------------------------------------------------------------
w = WorkspaceClient()

SECRET_SCOPE = "your_scope_name"

PG_HOST = "------------------.cloud.databricks.com"       
PG_PORT = 5432
PG_DBNAME = "-----------"

def get_decoded_secret(scope: str, key: str) -> str:
    """
    Fetch a secret from Databricks and return it as a clean, decoded string.
    Some SDK/workspace configs return secret .value as base64-encoded bytes-as-string
    rather than the plain original string, which silently breaks auth (garbled username/password).
    This tries a base64 decode first; if that fails or produces junk, it falls back to the raw value.
    """
    raw = w.secrets.get_secret(scope=scope, key=key).value
    try:
        decoded = base64.b64decode(raw).decode("utf-8")
        return decoded
    except Exception:
        # Not base64-encoded after all — use as-is
        return raw

PG_USER = get_decoded_secret(SECRET_SCOPE, "user")
PG_PASSWORD = get_decoded_secret(SECRET_SCOPE, "password")

# ---------------------------------------------------------------------------
# CONFIG: Views to refresh, IN ORDER.
# Each tuple is (schema-qualified view name, friendly label for logging)
# ---------------------------------------------------------------------------
VIEWS_TO_REFRESH = [
    ("insights.data_maturity", "Data Maturity"),
    ("insights.process_register", "Process Register"),
    ("insights.privacy_risk_readiness", "Privacy Risk Readiness"),
]

# ---------------------------------------------------------------------------
# ⏱️ CHANGE THIS to adjust the gap between refreshes (e.g. 30 for 30 minutes).
# ---------------------------------------------------------------------------
WAIT_MINUTES = 1

# COMMAND ----------

def get_connection():
    """Open a new SSL-secured connection to the Postgres/Lakebase database."""
    return psycopg2.connect(
        host=PG_HOST,
        port=PG_PORT,
        dbname=PG_DBNAME,
        user=PG_USER,
        password=PG_PASSWORD,
        sslmode="require",   # <-- Lakebase requires SSL; this fixes the "Invalid protocol version" error
    )


def refresh_view(view_name: str, label: str):
    """Run REFRESH MATERIALIZED VIEW for a single view, with basic error handling."""
    start = datetime.datetime.now()
    print(f"[{start.strftime('%Y-%m-%d %H:%M:%S')}] Starting refresh: {label} ({view_name})")

    conn = None
    try:
        conn = get_connection()
        conn.autocommit = True  # REFRESH MATERIALIZED VIEW cannot run inside a multi-statement transaction block
        with conn.cursor() as cur:
            cur.execute(f"REFRESH MATERIALIZED VIEW {view_name};")
        end = datetime.datetime.now()
        duration = (end - start).total_seconds()
        print(f"[{end.strftime('%Y-%m-%d %H:%M:%S')}] Finished refresh: {label} (took {duration:.1f}s)")
    except Exception as e:
        print(f"ERROR refreshing {label} ({view_name}): {e}")
        raise
    finally:
        if conn is not None:
            conn.close()

# COMMAND ----------

# ---------------------------------------------------------------------------
# Main sequence: refresh each view, waiting WAIT_MINUTES between each one.
# The wait happens AFTER a view finishes, before starting the next one.
# No wait after the last view in the list.
# ---------------------------------------------------------------------------
for i, (view_name, label) in enumerate(VIEWS_TO_REFRESH):
    refresh_view(view_name, label)

    is_last_view = (i == len(VIEWS_TO_REFRESH) - 1)
    if not is_last_view:
        # ⏱️ This is the 1-minute (WAIT_MINUTES) gap between refreshes.
        # Change WAIT_MINUTES above to adjust — no need to edit this block.
        print(f"Waiting {WAIT_MINUTES} minutes before next refresh...")
        time.sleep(WAIT_MINUTES * 60)

print("All materialized view refreshes complete.")

To schedule the workbook to run at required time:
1. Click on the right-hand side of the notebook, then click File > Schedule.
2. Click Add Schedule.
3. Add the appropriate information regarding the interval.
4. Click the Schedule tab and enter the timing information.
5. Click Create.

What's next

Generate dashboard reports via Databricks Lakebase

Generate dashboard reports via URL on Databricks Lakebase