databricks-data-handling

name	databricks-data-handling
description	Implement Delta Lake data management patterns including GDPR, PII handling, and data lifecycle. Use when implementing data retention, handling GDPR requests, or managing data lifecycle in Delta Lake. Trigger with phrases like "databricks GDPR", "databricks PII", "databricks data retention", "databricks data lifecycle", "delete user data".
allowed-tools	Read, Write, Edit, Bash(databricks:*)
version	1.0.0
license	MIT
author	Jeremy Longshore <[e-mail]>

Databricks Data Handling

Overview

Implement data management patterns for compliance, privacy, and lifecycle in Delta Lake.

Prerequisites

• Unity Catalog configured
• Understanding of Delta Lake features
• Compliance requirements documented
• Data classification in place

Instructions

Step 1: Data Classification and Tagging

-- Tag tables with data classification
ALTER TABLE catalog.schema.customers
SET TAGS ('data_classification' = 'PII', 'retention_days' = '365');

ALTER TABLE catalog.schema.orders
SET TAGS ('data_classification' = 'CONFIDENTIAL', 'retention_days' = '2555');

ALTER TABLE catalog.schema.analytics_events
SET TAGS ('data_classification' = 'INTERNAL', 'retention_days' = '90');

-- Tag columns with sensitivity
ALTER TABLE catalog.schema.customers
ALTER COLUMN email SET TAGS ('pii' = 'true', 'pii_type' = 'email');

ALTER TABLE catalog.schema.customers
ALTER COLUMN phone SET TAGS ('pii' = 'true', 'pii_type' = 'phone');

-- Query classified data
SELECT
    table_catalog,
    table_schema,
    table_name,
    tag_name,
    tag_value
FROM system.information_schema.table_tags
WHERE tag_name = 'data_classification';

Step 2: GDPR Right to Deletion (RTBF)

# src/compliance/gdpr.py
from pyspark.sql import SparkSession
from delta.tables import DeltaTable
from datetime import datetime
import logging

logger = logging.getLogger(__name__)

class GDPRHandler:
    """Handle GDPR data subject requests."""

    def __init__(self, spark: SparkSession, catalog: str):
        self.spark = spark
        self.catalog = catalog

    def process_deletion_request(
        self,
        user_id: str,
        request_id: str,
        dry_run: bool = True,
    ) -> dict:
        """
        Process GDPR deletion request for a user.

        Args:
            user_id: User identifier to delete
            request_id: GDPR request tracking ID
            dry_run: If True, only report what would be deleted

        Returns:
            Deletion report
        """
        report = {
            "request_id": request_id,
            "user_id": user_id,
            "timestamp": datetime.utcnow().isoformat(),
            "dry_run": dry_run,
            "tables_processed": [],
            "total_rows_deleted": 0,
        }

        # Find all tables with PII tag
        pii_tables = self._get_pii_tables()

        for table_info in pii_tables:
            table_name = f"{table_info['catalog']}.."
            user_column = ._get_user_column(table_name)

              user_column:
                

            
            count_query = 
            row_count = .spark.sql(count_query).first()[]

             row_count > :
                table_report = {
                    : table_name,
                    : row_count,
                    : ,
                }

                  dry_run:
                    
                    .spark.sql()
                    table_report[] = 

                    
                    ._log_deletion(request_id, table_name, user_id, row_count)

                report[].append(table_report)
                report[] += row_count

         report

     () -> []:
        
        query = 
         [row.asDict()  row  .spark.sql(query).collect()]

     () -> :
        
        
        columns = [c.name  c  .spark.table(table_name).schema]
        user_columns = [, , , ]

         uc  user_columns:
             uc  columns:
                 uc
         

     ():
        
        .spark.sql()


gdpr = GDPRHandler(spark, )
report = gdpr.process_deletion_request(
    user_id=,
    request_id=,
    dry_run=  
)
(report)

Step 3: Data Retention Policies

# src/compliance/retention.py
from pyspark.sql import SparkSession
from datetime import datetime, timedelta

class DataRetentionManager:
    """Manage data retention and cleanup."""

    def __init__(self, spark: SparkSession, catalog: str):
        self.spark = spark
        self.catalog = catalog

    def apply_retention_policies(self, dry_run: bool = True) -> list[dict]:
        """
        Apply retention policies based on table tags.

        Returns:
            List of tables processed with row counts
        """
        results = []

        # Get tables with retention tags
        tables = self.spark.sql(f"""
            SELECT
                table_catalog,
                table_schema,
                table_name,
                CAST(tag_value AS INT) as retention_days
            FROM {self.catalog}.information_schema.table_tags
            WHERE tag_name = 'retention_days'
        """).collect()

        for table in tables:
            full_name = f"{table.table_catalog}.{table.table_schema}.{table.table_name}"
            cutoff_date = datetime.now() - timedelta(days=table.retention_days)

            # Find date column
            date_col = self._get_date_column(full_name)
            if not date_col:
                continue

            # Count rows to delete
            count = .spark.sql().first()[]

            result = {
                : full_name,
                : table.retention_days,
                : cutoff_date.strftime(),
                : count,
                : ,
            }

              dry_run  count > :
                .spark.sql()
                result[] = 

            results.append(result)

         results

     () -> []:
        
        results = []

        
        tables = .spark.sql().collect()

         table  tables:
            full_name = 

            :
                .spark.sql()
                results.append({: full_name, : })
             Exception  e:
                results.append({: full_name, : , : (e)})

         results

     () -> :
        
        columns = [c.name  c  .spark.table(table_name).schema]
        date_columns = [, , , , ]

         dc  date_columns:
             dc  columns:
                 dc
         


 ():
    
    manager = DataRetentionManager(spark, )

    
    retention_results = manager.apply_retention_policies(dry_run=)
    ()

    
    vacuum_results = manager.vacuum_tables()
    ()

Step 4: PII Masking and Anonymization

# src/compliance/masking.py
from pyspark.sql import DataFrame
from pyspark.sql.functions import (
    col, sha2, concat, lit, regexp_replace,
    when, substring, length
)

class PIIMasker:
    """Mask PII data for analytics and testing."""

    @staticmethod
    def mask_email(df: DataFrame, column: str) -> DataFrame:
        """Mask email addresses: [e-mail] -> j***@***.com"""
        return df.withColumn(
            column,
            concat(
                substring(col(column), 1, 1),
                lit("***@***."),
                regexp_replace(col(column), r".*\.(\w+)$", "$1")
            )
        )

    @staticmethod
    def mask_phone(df: DataFrame, column: str) -> DataFrame:
        """Mask phone numbers: [phone] -> +1-555-***-****"""
        return df.withColumn(
            column,
            regexp_replace(col(column), r"(\d{3})-(\d{4})$", "***-****")
        )

    @staticmethod
    def hash_identifier(df: DataFrame, column: str, salt: str = "") -> DataFrame:
        """Hash identifiers for pseudonymization."""
        return df.withColumn(
            column,
            sha2(concat(col(column), lit(salt)), 256)
        )

    @staticmethod
    def () -> DataFrame:
        
         df.withColumn(
            column,
            regexp_replace(col(column), , )
        )


     () -> :
        
        df = spark.table(source_table)

         column, mask_type  masking_rules.items():
             mask_type == :
                df = PIIMasker.mask_email(df, column)
             mask_type == :
                df = PIIMasker.mask_phone(df, column)
             mask_type == :
                df = PIIMasker.hash_identifier(df, column)
             mask_type == :
                df = PIIMasker.mask_name(df, column)
             mask_type == :
                df = df.withColumn(column, lit())

        df.createOrReplaceTempView(view_name)


PIIMasker.create_masked_view(
    spark,
    ,
    ,
    {
        : ,
        : ,
        : ,
        : ,
    }
)

Step 5: Row-Level Security

-- Create row filter function
CREATE OR REPLACE FUNCTION catalog.security.region_filter(region STRING)
RETURNS BOOLEAN
RETURN (
    -- Allow access if user is admin
    IS_ACCOUNT_GROUP_MEMBER('data-admins')
    OR
    -- Or if region matches user's assigned region
    region = current_user_attribute('region')
);

-- Apply row filter to table
ALTER TABLE catalog.schema.sales
SET ROW FILTER catalog.security.region_filter ON (region);

-- Column masking function
CREATE OR REPLACE FUNCTION catalog.security.mask_salary(salary DECIMAL)
RETURNS DECIMAL
RETURN CASE
    WHEN IS_ACCOUNT_GROUP_MEMBER('hr-team') THEN salary
    ELSE NULL
END;

-- Apply column mask
ALTER TABLE catalog.schema.employees
ALTER COLUMN salary SET MASK catalog.security.mask_salary;

Output

• Data classification tags applied
• GDPR deletion process implemented
• Retention policies enforced
• PII masking configured
• Row-level security enabled

Error Handling

Issue	Cause	Solution
Vacuum fails	Retention too short	Ensure > 7 days retention
Delete timeout	Large table	Partition deletes over time
Missing user column	Non-standard schema	Map user columns manually
Mask function error	Invalid regex	Test masking functions

Examples

GDPR Subject Access Request

def generate_sar_report(spark, user_id: str) -> dict:
    """Generate Subject Access Request report."""
    pii_tables = get_pii_tables(spark)
    report = {"user_id": user_id, "data": {}}

    for table in pii_tables:
        user_col = get_user_column(table)
        if user_col:
            data = spark.sql(f"""
                SELECT * FROM {table}
                WHERE {user_col} = '{user_id}'
            """).toPandas().to_dict('records')
            report["data"][table] = data

    return report

Resources

• Delta Lake Privacy
• Unity Catalog Security
• GDPR Compliance

Next Steps

For enterprise RBAC, see databricks-enterprise-rbac.