🧱 Build · Explore · Optimise Databricks SQL

Upsert rows from a staging table into a Delta target. Inserts new rows, updates existing ones.

MERGE INTO catalog.schema.target_table AS tgt
USING (
  SELECT * FROM catalog.schema.staging_table
) AS src
ON tgt.id = src.id
WHEN MATCHED THEN
  UPDATE SET
    tgt.name       = src.name,
    tgt.value      = src.value,
    tgt.updated_at = current_timestamp()
WHEN NOT MATCHED T…

Track full history of dimension changes. Closes the old record and opens a new one on change.

MERGE INTO catalog.schema.dim_customer AS tgt
USING (
  SELECT
    src.customer_id,
    src.name,
    src.email,
    src.segment,
    current_timestamp() AS effective_from,
    CAST('9999-12-31' AS TIMESTAMP) AS effective_to,
    TRUE AS is_current,
    md5(concat_ws('|', src.nam…

Compact small files and co-locate related data for faster query pruning.

-- Compact small files in a partition range
OPTIMIZE catalog.schema.events_table
WHERE event_date >= '2025-01-01'
ZORDER BY (user_id, event_type);

-- Check file statistics after optimization
DESCRIBE DETAIL catalog.schema.events_table;

Delete files no longer referenced by the Delta log. Use DRY RUN first to preview.

-- Preview what will be deleted (safe — no actual deletion)
VACUUM catalog.schema.my_table DRY RUN;

-- Delete files older than 7 days (default retention)
VACUUM catalog.schema.my_table RETAIN 168 HOURS;

-- ⚠️  Short retention (disable safety check first)
SET spark.databricks.de…

Query a previous snapshot of a Delta table by version number.

-- View table history to find target version
DESCRIBE HISTORY catalog.schema.orders_table;

-- Query a specific version
SELECT * FROM catalog.schema.orders_table VERSION AS OF 42
WHERE order_status = 'PENDING';

-- Compare current vs historical count
SELECT
  (SELECT count(*) FRO…

Restore a point-in-time view using a timestamp string.

-- Query table as it was at a specific time
SELECT *
FROM catalog.schema.orders_table
TIMESTAMP AS OF '2025-09-22 00:00:00'
WHERE region = 'APAC';

-- Restore table to a previous state (overwrites current)
RESTORE TABLE catalog.schema.orders_table
TO TIMESTAMP AS OF '2025-09-22 0…

Create a shallow or deep clone of a Delta table for testing or archiving.

-- Shallow clone (references source files, instant)
CREATE TABLE catalog.schema.orders_clone_test
SHALLOW CLONE catalog.schema.orders_table
VERSION AS OF 100;

-- Deep clone (copies data files, fully independent)
CREATE TABLE catalog.schema.orders_archive_2024
DEEP CLONE catalog.…

Auto-managed clustering for faster queries — replaces manual Z-ORDER and partitioning.

-- Create table with Liquid Clustering
CREATE TABLE catalog.schema.sales (
  id           BIGINT,
  customer_id  BIGINT,
  product_id   BIGINT,
  sale_date    DATE,
  amount       DOUBLE,
  region       STRING
)
CLUSTER BY (customer_id, sale_date);

-- Add clustering to an existi…

Incrementally ingest new files from cloud storage using Databricks Auto Loader.

-- PySpark (run in a notebook cell)
-- Auto Loader: schema inference + checkpoint
df = (spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")           -- csv | parquet | avro | json
  .option("cloudFiles.schemaLocation", "abfss://container@storage.dfs.c…

Aggregate streaming events in time windows with late-data handling using watermarks.

-- Windowed aggregation on streaming data
df_windowed = (
  spark.readStream
    .format("delta")
    .table("catalog.schema.events_bronze")
    .withWatermark("event_time", "10 minutes")  -- tolerate 10 min late arrivals
    .groupBy(
      window("event_time", "5 minutes"),    …

Keep only the latest record per key using ROW_NUMBER in a CTE.

WITH ranked AS (
  SELECT
    *,
    ROW_NUMBER() OVER (
      PARTITION BY customer_id
      ORDER BY updated_at DESC
    ) AS rn
  FROM catalog.schema.raw_customers
)
SELECT * EXCEPT(rn)
FROM ranked
WHERE rn = 1;

Compare each row to the previous or next row within a partition.

SELECT
  order_id,
  customer_id,
  order_date,
  amount,
  LAG(amount,  1, 0) OVER (PARTITION BY customer_id ORDER BY order_date) AS prev_order_amt,
  LEAD(amount, 1, 0) OVER (PARTITION BY customer_id ORDER BY order_date) AS next_order_amt,
  amount - LAG(amount, 1, 0) OVER (PAR…

Compute cumulative sum and 7-day rolling average over ordered partitions.

SELECT
  sale_date,
  region,
  daily_revenue,
  SUM(daily_revenue) OVER (
    PARTITION BY region
    ORDER BY sale_date
    ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
  ) AS running_total,
  ROUND(AVG(daily_revenue) OVER (
    PARTITION BY region
    ORDER BY sale_date
  …

Get the first or last value in a window frame — useful for session analysis.

SELECT
  session_id,
  user_id,
  event_time,
  event_type,
  FIRST_VALUE(event_type) OVER (
    PARTITION BY session_id
    ORDER BY event_time
    ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING
  ) AS session_entry_event,
  LAST_VALUE(event_type) OVER (
    PARTITION …

Flatten an array column so each element becomes its own row.

-- Simple explode
SELECT
  order_id,
  explode(items) AS item
FROM catalog.schema.orders_with_arrays;

-- explode_outer preserves rows with null / empty arrays
SELECT
  user_id,
  posexplode_outer(tags) AS (tag_idx, tag)
FROM catalog.schema.user_profiles;

-- Inline explode of st…

Apply higher-order functions to arrays without exploding them.

SELECT
  user_id,
  tags,
  -- Transform: uppercase every tag
  transform(tags, t -> upper(t)) AS tags_upper,
  -- Filter: keep only tags starting with 'spark'
  filter(tags, t -> t LIKE 'spark%') AS spark_tags,
  -- Aggregate: concatenate array to string
  array_join(tags, ', ')…

Aggregate values into arrays, then manipulate them.

SELECT
  customer_id,
  collect_list(product_id)                              AS all_products_ordered,
  collect_set(product_id)                               AS unique_products,
  collect_list(DISTINCT product_id)                     AS distinct_ordered,
  size(collect_set(categ…

Infer schema from sample data, then parse a JSON string column into typed structs.

-- Step 1: infer schema from a sample row
SELECT schema_of_json(payload) AS inferred_schema
FROM catalog.schema.raw_events
LIMIT 1;

-- Step 2: parse using the inferred (or explicit) schema
SELECT
  event_id,
  event_time,
  from_json(
    payload,
    'STRUCT<user_id: BIGINT, ac…

Unnest deeply nested struct columns into a flat SELECT.

WITH parsed AS (
  SELECT
    event_id,
    event_time,
    from_json(payload, '
      STRUCT<
        user: STRUCT<id: BIGINT, name: STRING, country: STRING>,
        session: STRUCT<id: STRING, referrer: STRING, duration_s: INT>,
        items: ARRAY<STRUCT<sku: STRING, qty: IN…

Extract individual fields with get_json_object and serialize structs back to JSON.

SELECT
  event_id,
  -- Extract individual fields without defining a schema
  get_json_object(payload, '$.user.id')              AS user_id,
  get_json_object(payload, '$.session.duration_s')   AS duration,
  get_json_object(payload, '$.items[0].sku')         AS first_sku,

  -- …

Chain multiple CTEs for readable, step-by-step data transformation.

WITH
-- Stage 1: filter and clean raw data
cleaned AS (
  SELECT
    order_id,
    customer_id,
    product_id,
    COALESCE(amount, 0.0)   AS amount,
    CAST(order_date AS DATE) AS order_date,
    TRIM(LOWER(region))      AS region
  FROM catalog.schema.raw_orders
  WHERE order…

Rotate rows to columns (PIVOT) and columns to rows (UNPIVOT).

-- PIVOT: quarters as columns
SELECT *
FROM (
  SELECT
    region,
    CONCAT('Q', EXTRACT(quarter FROM sale_date)) AS quarter,
    amount
  FROM catalog.schema.sales
)
PIVOT (
  ROUND(SUM(amount), 2) AS revenue
  FOR quarter IN ('Q1', 'Q2', 'Q3', 'Q4')
);

-- UNPIVOT: turn metri…

Compute multiple levels of aggregation in a single pass.

-- GROUPING SETS: explicit combinations
SELECT
  COALESCE(region,   'ALL REGIONS')  AS region,
  COALESCE(category, 'ALL CATS')     AS category,
  COALESCE(brand,    'ALL BRANDS')   AS brand,
  SUM(revenue) AS total_revenue,
  GROUPING(region)   AS is_region_total,
  GROUPING(cat…

Use approximate aggregations for speed on billion-row tables.

SELECT
  region,
  date_trunc('month', event_date)        AS month,
  COUNT(*)                               AS total_events,
  -- Approximate distinct count (much faster than COUNT(DISTINCT ...))
  approx_count_distinct(user_id)         AS approx_unique_users,
  approx_count_dis…

Manage fine-grained access control with Unity Catalog GRANT statements.

-- Grant catalog-level access
GRANT USE CATALOG ON CATALOG my_catalog TO `data-team@company.com`;

-- Grant schema access
GRANT USE SCHEMA, CREATE TABLE ON SCHEMA my_catalog.analytics
  TO `analysts-group`;

-- Grant table-level SELECT
GRANT SELECT ON TABLE my_catalog.analytics.f…

Explore Unity Catalog metadata: tables, columns, lineage, tags.

-- List all tables in a schema with row count estimate
SELECT
  table_catalog,
  table_schema,
  table_name,
  table_type,
  created,
  last_altered
FROM my_catalog.information_schema.tables
WHERE table_schema = 'analytics'
ORDER BY last_altered DESC;

-- Find all columns matchin…

Create external Delta tables and access files via Unity Catalog Volumes.

-- Create external table pointing to cloud storage
CREATE TABLE my_catalog.raw.external_logs
USING DELTA
LOCATION 'abfss://container@storageaccount.dfs.core.windows.net/logs/'
TBLPROPERTIES (
  'delta.columnMapping.mode' = 'name',
  'delta.enableDeletionVectors' = 'true'
);

-- C…

Control join strategies with hints to avoid shuffle-heavy sort-merge joins.

-- BROADCAST: force small table to be broadcast (no shuffle)
SELECT /*+ BROADCAST(d) */
  f.order_id,
  f.customer_id,
  d.segment,
  d.country,
  f.amount
FROM catalog.schema.fact_orders f
JOIN catalog.schema.dim_customer d   -- small dim table
  ON f.customer_id = d.customer_id…

Control output file count and partition layout to optimize read performance.

-- Write with controlled partition count (reduce small files)
-- PySpark
df.coalesce(8).write.format("delta").mode("overwrite")   .partitionBy("sale_date")   .save("abfss://...")

-- Repartition by key for balanced shuffle output
df.repartition(200, "customer_id", "region")   .wr…

Cache hot tables, persist intermediates, and read EXPLAIN plans.

-- Cache a frequently-joined dimension in memory
CACHE TABLE catalog.schema.dim_product;
CACHE LAZY TABLE catalog.schema.dim_customer;  -- cached on first use

-- Uncache when done
UNCACHE TABLE catalog.schema.dim_product;

-- Delta Cache (auto on Databricks, toggle per cluster):…

Call GPT-4o or Claude directly from a SQL query using Databricks AI Functions.

-- Sentiment analysis on review text
SELECT
  review_id,
  review_text,
  ai_query(
    'databricks-meta-llama-3-1-70b-instruct',
    CONCAT(
      'Classify the sentiment of this product review as POSITIVE, NEGATIVE, or NEUTRAL. ',
      'Respond with only the label. Review: ', …

Score a table with a registered MLflow model using the predict() SQL function.

-- Register model first via MLflow (Python):
-- mlflow.register_model("runs:/abc123/model", "catalog.ml.churn_model")

-- Batch scoring: score all customers in a single SQL query
SELECT
  c.customer_id,
  c.segment,
  c.tenure_months,
  c.monthly_spend,
  c.support_tickets_90d,
 …

Create a managed Delta table with partitioning, constraints, and table properties.

CREATE TABLE IF NOT EXISTS catalog.schema.fact_sales (
  sale_id          BIGINT        NOT NULL,
  customer_id      BIGINT        NOT NULL,
  product_id       BIGINT        NOT NULL,
  sale_date        DATE          NOT NULL,
  region           STRING        NOT NULL,
  amount  …

Load files from cloud storage into a Delta table incrementally using COPY INTO.

-- Create target table first
CREATE TABLE IF NOT EXISTS catalog.raw.orders_bronze
USING DELTA
LOCATION 'abfss://container@storage.dfs.core.windows.net/delta/orders_bronze/';

-- COPY INTO: idempotent incremental load (tracks already-loaded files)
COPY INTO catalog.raw.orders_bron…

Add, rename, drop columns and change Delta table properties safely.

-- Add columns (Delta supports schema evolution)
ALTER TABLE catalog.schema.fact_sales
  ADD COLUMNS (
    campaign_id  STRING   COMMENT 'Marketing campaign ID',
    channel      STRING   COMMENT 'Acquisition channel: web | app | store',
    is_returned  BOOLEAN  DEFAULT false
  …