Metastore

• Top-level container: Bevat alle metadata
• Cross-workspace: Gedeeld tussen meerdere workspaces
• Central management: Eén beheerpunt voor governance
• Backup/restore: Metadata backup mogelijkheden

Catalog

• Eerste organisatie niveau: Bijv. departementen
• Isolatie boundary: Security en access boundary
• Resource grouping: Logische groepering van data
• Default: "main": Standaard catalog

Schema

• Database equivalent: Container voor tables/views
• Data lake mapping: Mapt naar directories
• Medallion architecture: bronze/silver/gold
• Access control: Permissions op schema niveau

Lineage Tracking

-- SQL: Lineage queries
-- Volledige upstream lineage
SELECT * FROM system.information_schema.table_lineage
WHERE downstream_catalog = 'main'
  AND downstream_schema = 'gold'
  AND downstream_table = 'customer_360';

-- Downstream impact analysis
SELECT * FROM system.information_schema.column_lineage
WHERE upstream_catalog = 'main'
  AND upstream_schema = 'silver'
  AND upstream_table = 'transactions'
  AND upstream_column = 'amount';

-- Lineage visualisatie in notebook
%python
from databricks.sdk import WorkspaceClient
import networkx as nx

def visualize_lineage(table_name):
    w = WorkspaceClient()
    lineage = w.data_lineage.get(table_name)
    
    # Build graph
    G = nx.DiGraph()
    for edge in lineage.edges:
        G.add_edge(edge.source, edge.target)
    
    # Visualize
    nx.draw(G, with_labels=True)
    return G

Tagging & Classification

-- SQL: Tagging en classificatie
-- Maak custom tags
CREATE TAG IF NOT EXISTS data_sensitivity 
COMMENT 'Data sensitivity classification';

CREATE TAG IF NOT EXISTS data_quality 
COMMENT 'Data quality metrics';

CREATE TAG IF NOT EXISTS business_domain 
COMMENT 'Business domain classification';

-- Pas tags toe op tabellen
ALTER TABLE main.sensitive.customers 
SET TAG data_sensitivity = 'CONFIDENTIAL';

ALTER TABLE main.gold.sales 
SET TAG business_domain = 'SALES',
    TAG data_quality = 'CERTIFIED';

-- Pas tags toe op kolommen (PII detectie)
ALTER TABLE main.sensitive.customers 
ALTER COLUMN email 
SET TAG data_sensitivity = 'PII';

ALTER TABLE main.sensitive.customers 
ALTER COLUMN bsn 
SET TAG data_sensitivity = 'SENSITIVE_PII';

-- Query op basis van tags
SELECT 
  table_catalog,
  table_schema,
  table_name,
  tag_name,
  tag_value
FROM system.information_schema.table_tags
WHERE tag_name = 'data_sensitivity'
  AND tag_value = 'PII';

Fine-grained Access Control

-- SQL: Advanced access control
-- Column-level security
CREATE MASKING POLICY email_mask AS (email STRING) 
RETURN CASE 
  WHEN current_user() IN ('admin', 'data_engineer') 
    THEN email
  ELSE regexp_replace(email, '(?<=.).(?=.*@)', '*')
END;

CREATE MASKING POLICY amount_mask AS (amount DECIMAL) 
RETURN CASE 
  WHEN current_user() IN ('admin', 'executive') 
    THEN amount
  ELSE NULL
END;

-- Pas masking policies toe
ALTER TABLE main.sensitive.customers 
ALTER COLUMN email 
SET MASKING POLICY email_mask;

ALTER TABLE main.gold.sales 
ALTER COLUMN revenue 
SET MASKING POLICY amount_mask;

-- Row-level security (filter)
CREATE ROW FILTER ssn_filter AS (ssn STRING) 
RETURN CASE 
  WHEN current_user() IN ('admin', 'hr_user') 
    THEN true
  ELSE false
END;

ALTER TABLE main.sensitive.employees 
SET ROW FILTER ssn_filter ON (ssn);

-- Dynamic views met security
CREATE VIEW main.analytics.safe_customer_data AS
SELECT 
  customer_id,
  CASE 
    WHEN current_user() IN ('admin', 'sales_manager') 
      THEN customer_name
    ELSE 'CONFIDENTIAL'
  END as customer_name,
  CASE 
    WHEN current_user() IN ('admin', 'sales_manager') 
      THEN email
    ELSE regexp_replace(email, '(?<=.).(?=.*@)', '*')
  END as email,
  country,
  customer_segment
FROM main.sensitive.customers;

Data Sharing & Collaboration

-- SQL: Data sharing binnen en buiten organisatie
-- INTERN: Share binnen metastore
CREATE SHARE sales_data_share
COMMENT 'Share sales data with analytics team';

-- Voeg objecten toe aan share
ALTER SHARE sales_data_share 
ADD TABLE main.gold.sales;

ALTER SHARE sales_data_share 
ADD VIEW main.analytics.sales_dashboard;

-- Grant access aan andere catalogs
GRANT SELECT ON SHARE sales_data_share 
TO CATALOG analytics;

-- EXTERN: Delta Sharing (open protocol)
-- Maak een provider (als je data deelt)
CREATE PROVIDER external_partner 
COMMENT 'External business partner';

-- Maak een recipient (als je data ontvangt)
CREATE RECIPIENT supplier_data
COMMENT 'Supplier data share';

-- Deel data via Delta Sharing
CREATE SHARE external_data_share AS
SELECT * FROM main.gold.shared_metrics;

GRANT SELECT ON SHARE external_data_share 
TO RECIPIENT supplier_data;

-- Query gedeelde data (als recipient)
SELECT * FROM delta_sharing.supplier_data.sales_metrics;

-- Real-time data sharing
CREATE LIVE TABLE real_time_metrics
AS SELECT 
  current_timestamp() as timestamp,
  count(*) as active_sessions,
  avg(response_time) as avg_response_time
FROM system.streaming.sessions
GROUP BY window(timestamp, '1 minute');

-- Share live table
ALTER SHARE real_time_share 
ADD LIVE TABLE real_time_metrics;

Query Performance

-- SQL: Performance monitoring en optimalisatie
-- 1. Query performance analysis
CREATE OR REPLACE TABLE main.monitoring.query_performance
AS SELECT 
  query_id,
  query_text,
  user_name,
  execution_time,
  result_size,
  rows_produced,
  -- Performance metrics
  execution_time / rows_produced as ms_per_row,
  CASE 
    WHEN execution_time > 60000 THEN 'SLOW'
    WHEN execution_time > 30000 THEN 'MEDIUM'
    ELSE 'FAST'
  END as performance_category
FROM system.query.history
WHERE start_time > current_date() - 7;

-- 2. Table statistics voor query optimizer
ANALYZE TABLE main.gold.sales 
COMPUTE STATISTICS FOR ALL COLUMNS;

-- Show table statistics
DESCRIBE DETAIL main.gold.sales;

-- 3. Z-Porder optimalisatie (co-locatie)
CREATE OR REPLACE TABLE main.gold.optimized_sales
USING DELTA
LOCATION 's3://company-data/gold/sales/'
TBLPROPERTIES (
  'delta.autoOptimize.optimizeWrite' = 'true',
  'delta.autoOptimize.autoCompact' = 'true'
)
AS SELECT * FROM main.silver.sales
-- Optimaliseer voor meest gebruikte queries
CLUSTER BY (customer_id, order_date);

-- 4. Materialized views voor vaak gebruikte aggregaties
CREATE MATERIALIZED VIEW main.gold.daily_sales_mv
REFRESH EVERY 1 HOUR
AS SELECT 
  date_trunc('day', order_date) as sales_day,
  customer_segment,
  product_category,
  SUM(order_amount) as daily_sales,
  COUNT(DISTINCT customer_id) as unique_customers,
  AVG(order_amount) as avg_order_value
FROM main.gold.sales
GROUP BY 1, 2, 3;

-- 5. Query acceleration met Photon
SET use_photon = true;

-- Force Photon voor specifieke query
SELECT /*+ photon */ 
  customer_id,
  SUM(order_amount) as total_spent
FROM main.gold.sales
GROUP BY customer_id
ORDER BY total_spent DESC
LIMIT 100;

Storage Optimalisatie

-- SQL: Storage management en optimalisatie
-- 1. Storage analytics
CREATE OR REPLACE VIEW main.monitoring.storage_usage AS
SELECT 
  catalog_name,
  schema_name,
  table_name,
  -- Storage metrics
  SUM(size_in_bytes) / 1024 / 1024 / 1024 as size_gb,
  SUM(num_files) as total_files,
  AVG(file_size) as avg_file_size_mb,
  -- Optimalisatie opportunities
  CASE 
    WHEN AVG(file_size) < 100 * 1024 * 1024 THEN 'SMALL_FILES'
    WHEN COUNT(*) > 1000 THEN 'TOO_MANY_FILES'
    ELSE 'OPTIMAL'
  END as optimization_status
FROM system.information_schema.table_storage
GROUP BY 1, 2, 3
ORDER BY size_gb DESC;

-- 2. Auto-optimization configuratie
ALTER TABLE main.gold.large_table 
SET TBLPROPERTIES (
  'delta.autoOptimize.optimizeWrite' = 'true',
  'delta.autoOptimize.autoCompact' = 'true',
  'delta.targetFileSize' = '128mb'
);

-- 3. Manual optimization
OPTIMIZE main.gold.sales 
ZORDER BY (customer_id, order_date);

-- Show optimization results
DESCRIBE HISTORY main.gold.sales;

-- 4. Vacuum (clean up old versions)
VACUUM main.gold.sales 
RETAIN 168 HOURS; -- 7 dagen retention

-- 5. Clone voor testing zonder data kopiëren
-- Zero-copy clone (metadata only)
CREATE TABLE main.sandbox.sales_test
CLONE main.gold.sales;

-- Deep clone (copy data)
CREATE TABLE main.backup.sales_backup
DEEP CLONE main.gold.sales;

-- 6. Table maintenance procedure
CREATE OR REPLACE PROCEDURE maintain_tables()
LANGUAGE SQL
AS $$
DECLARE
  cursor CURSOR FOR 
    SELECT 
      table_catalog,
      table_schema,
      table_name
    FROM system.information_schema.tables
    WHERE table_schema IN ('bronze', 'silver', 'gold')
      AND table_type = 'BASE TABLE';
BEGIN
  FOR record IN cursor LOOP
    -- Optimize tables groter dan 1GB
    EXECUTE IMMEDIATE format('
      OPTIMIZE %I.%I.%I 
      WHERE size_in_bytes > 1073741824',
      record.table_catalog,
      record.table_schema,
      record.table_name
    );
    
    -- Vacuum oude versies
    EXECUTE IMMEDIATE format('
      VACUUM %I.%I.%I 
      RETAIN 168 HOURS',
      record.table_catalog,
      record.table_schema,
      record.table_name
    );
  END LOOP;
END;
$$;

Kosten Breakdown

-- SQL: Kosten monitoring en optimalisatie
-- 1. Kosten tracking per team/department
CREATE OR REPLACE VIEW main.finance.cost_analysis AS
WITH compute_costs AS (
  SELECT 
    user_name,
    -- Extract team from email/username
    SPLIT(user_name, '@')[0] as team,
    SUM(DBUs_used) as total_dbus,
    SUM(DBUs_used * 0.65) as compute_cost, -- €0.65/DBU
    COUNT(DISTINCT query_id) as query_count,
    AVG(execution_time) / 1000 as avg_query_seconds
  FROM system.billing.usage
  WHERE usage_date >= DATE_TRUNC('month', CURRENT_DATE())
  GROUP BY 1
),
storage_costs AS (
  SELECT 
    table_catalog,
    table_schema,
    -- Map schema to team
    CASE 
      WHEN table_schema LIKE '%sales%' THEN 'sales'
      WHEN table_schema LIKE '%marketing%' THEN 'marketing'
      WHEN table_schema LIKE '%finance%' THEN 'finance'
      ELSE 'other'
    END as team,
    SUM(size_in_bytes) / 1024 / 1024 / 1024 as storage_gb,
    SUM(size_in_bytes) / 1024 / 1024 / 1024 * 0.023 as storage_cost -- €0.023/GB/maand
  FROM system.information_schema.table_storage
  GROUP BY 1, 2
),
unity_catalog_costs AS (
  SELECT 
    'unity_catalog' as cost_type,
    COUNT(DISTINCT table_name) * 0.20 as catalog_cost -- €0.20/table/maand
  FROM system.information_schema.tables
)
SELECT 
  COALESCE(cc.team, sc.team) as team,
  SUM(COALESCE(cc.compute_cost, 0)) as compute_cost,
  SUM(COALESCE(sc.storage_cost, 0)) as storage_cost,
  SUM(COALESCE(cc.compute_cost, 0) + COALESCE(sc.storage_cost, 0)) as total_cost,
  -- Cost efficiency metrics
  SUM(COALESCE(cc.query_count, 0)) / 
    NULLIF(SUM(COALESCE(cc.compute_cost, 0)), 0) as queries_per_euro,
  SUM(COALESCE(sc.storage_gb, 0)) / 
    NULLIF(SUM(COALESCE(cc.query_count, 0)), 0) as gb_per_query
FROM compute_costs cc
FULL OUTER JOIN storage_costs sc
  ON cc.team = sc.team
GROUP BY 1
ORDER BY total_cost DESC;

-- 2. Kosten optimalisatie recommendations
CREATE OR REPLACE VIEW main.finance.cost_optimization AS
SELECT 
  'STORAGE' as category,
  table_name,
  size_gb,
  CASE 
    WHEN size_gb > 1000 THEN 'CONSIDER ARCHIVING'
    WHEN last_access_days > 90 THEN 'CONSIDER DELETION'
    WHEN file_count > 10000 THEN 'CONSIDER OPTIMIZE'
    ELSE 'OPTIMAL'
  END as recommendation,
  estimated_savings
FROM (
  SELECT 
    table_name,
    SUM(size_in_bytes) / 1024 / 1024 / 1024 as size_gb,
    COUNT(*) as file_count,
    DATEDIFF(day, MAX(last_access), CURRENT_DATE()) as last_access_days,
    CASE 
      WHEN DATEDIFF(day, MAX(last_access), CURRENT_DATE()) > 365 
        THEN SUM(size_in_bytes) / 1024 / 1024 / 1024 * 0.023 * 12
      ELSE 0
    END as estimated_savings
  FROM system.information_schema.table_storage
  GROUP BY 1
)
UNION ALL
SELECT 
  'COMPUTE' as category,
  user_name as resource,
  total_dbus,
  CASE 
    WHEN avg_query_time > 30000 THEN 'OPTIMIZE QUERIES'
    WHEN query_count < 10 THEN 'REDUCE ENVIRONMENT SIZE'
    ELSE 'OPTIMAL'
  END as recommendation,
  CASE 
    WHEN avg_query_time > 30000 THEN total_dbus * 0.10
    ELSE 0
  END as estimated_savings
FROM (
  SELECT 
    user_name,
    SUM(DBUs_used) as total_dbus,
    AVG(execution_time) as avg_query_time,
    COUNT(*) as query_count
  FROM system.billing.usage
  GROUP BY 1
);

-- 3. Auto-scale recommendations
CREATE OR REPLACE PROCEDURE optimize_cluster_sizes()
LANGUAGE SQL
AS $$
DECLARE
  avg_utilization DECIMAL;
  peak_utilization DECIMAL;
BEGIN
  -- Calculate utilization
  SELECT 
    AVG(active_workers / max_workers),
    MAX(active_workers / max_workers)
  INTO avg_utilization, peak_utilization
  FROM system.cluster.usage
  WHERE timestamp > CURRENT_TIMESTAMP() - INTERVAL '7 days';
  
  -- Make recommendations
  IF avg_utilization < 0.3 THEN
    CALL system.alert('UNDERUTILIZED_CLUSTERS',
      'Average cluster utilization only ' || ROUND(avg_utilization * 100, 1) || '%. Consider downsizing.');
  ELSIF peak_utilization > 0.9 THEN
    CALL system.alert('OVERUTILIZED_CLUSTERS',
      'Peak cluster utilization ' || ROUND(peak_utilization * 100, 1) || '%. Consider upsizing.');
  END IF;
END;
$$;