Hive to dbt / SparkSQL Migration¶

-- After data migration to ADLS, repair partition metadata
-- In Databricks (Unity Catalog)
MSCK REPAIR TABLE silver.orders;

-- Or for Delta tables, partitions are discovered automatically
-- No MSCK REPAIR needed if using CONVERT TO DELTA

-- BEFORE: HiveQL
SET hive.exec.parallel=true;
SET hive.exec.dynamic.partition.mode=nonstrict;
SET hive.exec.max.dynamic.partitions=10000;

INSERT OVERWRITE TABLE analytics.daily_orders
PARTITION (order_date)
SELECT
    customer_id,
    COUNT(*) AS order_count,
    SUM(amount) AS total_amount,
    AVG(amount) AS avg_amount,
    order_date
FROM raw.orders
WHERE order_date >= '${hiveconf:start_date}'
GROUP BY customer_id, order_date;

-- AFTER: SparkSQL (Databricks)
INSERT OVERWRITE TABLE silver.daily_orders
SELECT
    customer_id,
    COUNT(*) AS order_count,
    SUM(amount) AS total_amount,
    AVG(amount) AS avg_amount,
    order_date
FROM bronze.orders
WHERE order_date >= :start_date
GROUP BY customer_id, order_date;

-- BEFORE: HiveQL
CREATE TABLE IF NOT EXISTS analytics.customer_360
STORED AS ORC
AS
SELECT
    c.customer_id,
    c.name,
    c.segment,
    o.order_count,
    o.total_revenue,
    p.last_payment_date,
    COALESCE(s.support_tickets, 0) AS support_tickets,
    CASE
        WHEN o.total_revenue > 10000 THEN 'high'
        WHEN o.total_revenue > 1000 THEN 'medium'
        ELSE 'low'
    END AS value_tier
FROM raw.customers c
LEFT JOIN (
    SELECT customer_id, COUNT(*) AS order_count, SUM(amount) AS total_revenue
    FROM raw.orders
    GROUP BY customer_id
) o ON c.customer_id = o.customer_id
LEFT JOIN (
    SELECT customer_id, MAX(payment_date) AS last_payment_date
    FROM raw.payments
    GROUP BY customer_id
) p ON c.customer_id = p.customer_id
LEFT JOIN (
    SELECT customer_id, COUNT(*) AS support_tickets
    FROM raw.support
    GROUP BY customer_id
) s ON c.customer_id = s.customer_id;

-- AFTER: SparkSQL / Delta Lake
CREATE OR REPLACE TABLE gold.customer_360
USING DELTA
AS
SELECT
    c.customer_id,
    c.name,
    c.segment,
    o.order_count,
    o.total_revenue,
    p.last_payment_date,
    COALESCE(s.support_tickets, 0) AS support_tickets,
    CASE
        WHEN o.total_revenue > 10000 THEN 'high'
        WHEN o.total_revenue > 1000 THEN 'medium'
        ELSE 'low'
    END AS value_tier
FROM silver.customers c
LEFT JOIN (
    SELECT customer_id, COUNT(*) AS order_count, SUM(amount) AS total_revenue
    FROM silver.orders
    GROUP BY customer_id
) o ON c.customer_id = o.customer_id
LEFT JOIN (
    SELECT customer_id, MAX(payment_date) AS last_payment_date
    FROM silver.payments
    GROUP BY customer_id
) p ON c.customer_id = p.customer_id
LEFT JOIN (
    SELECT customer_id, COUNT(*) AS support_tickets
    FROM silver.support
    GROUP BY customer_id
) s ON c.customer_id = s.customer_id;

-- Hive bucketed table
CREATE TABLE orders_bucketed (
    order_id BIGINT,
    customer_id BIGINT,
    amount DECIMAL(10,2),
    order_date DATE
)
CLUSTERED BY (customer_id) INTO 256 BUCKETS
STORED AS ORC;

-- Delta table with Z-ORDER optimization
CREATE TABLE silver.orders (
    order_id BIGINT,
    customer_id BIGINT,
    amount DECIMAL(10,2),
    order_date DATE
)
USING DELTA;

-- Run Z-ORDER after data load
OPTIMIZE silver.orders ZORDER BY (customer_id);

-- Liquid clustering (auto-managed, no manual OPTIMIZE needed)
CREATE TABLE silver.orders (
    order_id BIGINT,
    customer_id BIGINT,
    amount DECIMAL(10,2),
    order_date DATE
)
USING DELTA
CLUSTER BY (customer_id, order_date);

-- Hive table with custom SerDe for JSON data
CREATE EXTERNAL TABLE raw.events (
    event_id STRING,
    event_type STRING,
    payload STRING
)
ROW FORMAT SERDE 'org.apache.hive.hcatalog.data.JsonSerDe'
STORED AS TEXTFILE
LOCATION 'hdfs:///data/events/';

-- Delta table with native JSON support
-- Step 1: Read JSON and write as Delta
-- (in a Databricks notebook)

-- Read JSON files natively
CREATE TABLE bronze.events
USING DELTA
AS SELECT * FROM json.`abfss://raw@storage.dfs.core.windows.net/events/`;

-- Step 2: Schema evolution is automatic
-- New fields in JSON files are added automatically with:
ALTER TABLE bronze.events SET TBLPROPERTIES ('delta.columnMapping.mode' = 'name');

-- Static partition insert
INSERT INTO TABLE orders PARTITION (year=2025, month=04)
SELECT order_id, customer_id, amount FROM staging_orders;

-- Dynamic partition insert
INSERT INTO TABLE orders PARTITION (year, month)
SELECT order_id, customer_id, amount, year, month FROM staging_orders;

-- Delta supports the same partition semantics
INSERT INTO silver.orders
SELECT order_id, customer_id, amount, year, month FROM bronze.orders;
-- Delta automatically writes to partition directories based on partitionBy columns

-- models/silver/daily_orders.sql

{{ config(
    materialized='incremental',
    unique_key='customer_id || order_date',
    partition_by=['order_date'],
    file_format='delta'
) }}

SELECT
    customer_id,
    COUNT(*) AS order_count,
    SUM(amount) AS total_amount,
    AVG(amount) AS avg_amount,
    order_date
FROM {{ ref('stg_orders') }}

{% if is_incremental() %}
WHERE order_date > (SELECT MAX(order_date) FROM {{ this }})
{% endif %}

GROUP BY customer_id, order_date